diff --git a/.travis/build.sh b/.travis/build.sh
index 0f7e9930..8851adbd 100755
--- a/.travis/build.sh
+++ b/.travis/build.sh
@@ -27,7 +27,13 @@ mvn spotbugs:check
 # Run testsuite with java 8 only
 if [ ${JAVA_MAJOR_VERSION} -eq 1 ] ; then
   mvn test-compile spotbugs:check -e -V -B -f testsuite
+  set +e
   mvn -e -V -B install -f testsuite
+
+  EXIT=$?
+  FILE=testsuite/kafka.log
+  [ "$EXIT" != "0" ] && test -f "$FILE" && cat $FILE && exit $EXIT
+  set -e
 fi
 
 # Push only releases
diff --git a/oauth-common/pom.xml b/oauth-common/pom.xml
index af0b1287..452c6c43 100644
--- a/oauth-common/pom.xml
+++ b/oauth-common/pom.xml
@@ -16,6 +16,14 @@
             <groupId>org.keycloak</groupId>
             <artifactId>keycloak-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.keycloak</groupId>
+            <artifactId>keycloak-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-jdk15on</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/Config.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/Config.java
index 469b698e..8b313e2b 100644
--- a/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/Config.java
+++ b/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/Config.java
@@ -130,8 +130,7 @@ public boolean getValueAsBoolean(String key, boolean fallback) {
 
     private boolean isTrue(String result) {
         String val = result.toLowerCase(Locale.ENGLISH);
-        boolean tru = val.equals("true") || val.equals("yes") || val.equals("y") || val.equals("1");
-        if (true) {
+        if (val.equals("true") || val.equals("yes") || val.equals("y") || val.equals("1")) {
             return true;
         }
         if (val.equals("false") || val.equals("no") || val.equals("n") || val.equals("0")) {
diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/ECDSASignatureVerifierContext.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/ECDSASignatureVerifierContext.java
new file mode 100644
index 00000000..0e1281d0
--- /dev/null
+++ b/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/ECDSASignatureVerifierContext.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2017-2020, Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.oauth.validator;
+
+import org.bouncycastle.asn1.ASN1Integer;
+import org.bouncycastle.asn1.DERSequenceGenerator;
+import org.keycloak.common.VerificationException;
+import org.keycloak.crypto.AsymmetricSignatureVerifierContext;
+import org.keycloak.crypto.KeyWrapper;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigInteger;
+
+/**
+ * This class provides ECDSA signature verification support.
+ *
+ * Adapted from: https://github.com/keycloak/keycloak/blob/8.0.1/services/src/main/java/org/keycloak/crypto/ClientECDSASignatureVerifierContext.java
+ */
+public class ECDSASignatureVerifierContext extends AsymmetricSignatureVerifierContext {
+
+    public ECDSASignatureVerifierContext(KeyWrapper key) {
+        super(key);
+    }
+
+    @Override
+    public boolean verify(byte[] data, byte[] signature) throws VerificationException {
+       /*
+       Fallback for backwards compatibility of ECDSA signed tokens which were issued in previous versions.
+       TODO remove by https://issues.jboss.org/browse/KEYCLOAK-11911
+        */
+        int expectedSize = ECDSA.valueOf(getAlgorithm()).getSignatureLength();
+        byte[] derSignature = expectedSize != signature.length && signature[0] == 0x30 ? signature : concatenatedRSToASN1DER(signature, expectedSize);
+        return super.verify(data, derSignature);
+    }
+
+    enum ECDSA {
+        ES256(64),
+        ES384(96),
+        ES512(132);
+
+        private final int signatureLength;
+
+        ECDSA(int signatureLength) {
+            this.signatureLength = signatureLength;
+        }
+
+        public int getSignatureLength() {
+            return this.signatureLength;
+        }
+    }
+
+    static byte[] concatenatedRSToASN1DER(final byte[] signature, int signLength) {
+        int len = signLength / 2;
+        int arraySize = len + 1;
+
+        byte[] r = new byte[arraySize];
+        byte[] s = new byte[arraySize];
+        System.arraycopy(signature, 0, r, 1, len);
+        System.arraycopy(signature, len, s, 1, len);
+        BigInteger rBigInteger = new BigInteger(r);
+        BigInteger sBigInteger = new BigInteger(s);
+
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        try {
+            DERSequenceGenerator seqGen = new DERSequenceGenerator(bos);
+
+            seqGen.addObject(new ASN1Integer(rBigInteger.toByteArray()));
+            seqGen.addObject(new ASN1Integer(sBigInteger.toByteArray()));
+            seqGen.close();
+            bos.close();
+        } catch (IOException e) {
+            throw new RuntimeException("Failed to generate ASN.1 DER signature", e);
+        }
+        return bos.toByteArray();
+    }
+}
\ No newline at end of file
diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/JWTSignatureValidator.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/JWTSignatureValidator.java
index 255a69b9..12ae20ff 100644
--- a/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/JWTSignatureValidator.java
+++ b/oauth-common/src/main/java/io/strimzi/kafka/oauth/validator/JWTSignatureValidator.java
@@ -9,7 +9,10 @@
 import io.strimzi.kafka.oauth.common.TimeUtil;
 import io.strimzi.kafka.oauth.common.TokenInfo;
 import org.apache.kafka.common.utils.Time;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
 import org.keycloak.TokenVerifier;
+import org.keycloak.crypto.AsymmetricSignatureVerifierContext;
+import org.keycloak.crypto.KeyWrapper;
 import org.keycloak.exceptions.TokenSignatureInvalidException;
 import org.keycloak.jose.jwk.JSONWebKeySet;
 import org.keycloak.jose.jwk.JWK;
@@ -22,13 +25,16 @@
 import javax.net.ssl.SSLSocketFactory;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.security.Provider;
 import java.security.PublicKey;
+import java.security.Security;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static io.strimzi.kafka.oauth.validator.TokenValidationException.Status;
 import static org.keycloak.TokenVerifier.IS_ACTIVE;
@@ -38,6 +44,8 @@ public class JWTSignatureValidator implements TokenValidator {
 
     private static final Logger log = LoggerFactory.getLogger(JWTSignatureValidator.class);
 
+    private static AtomicBoolean bouncyInstalled =  new AtomicBoolean(false);
+
     private final ScheduledExecutorService scheduler;
 
     private final URI keysUri;
@@ -62,7 +70,9 @@ public JWTSignatureValidator(String keysEndpointUri,
                                  int expirySeconds,
                                  boolean defaultChecks,
                                  boolean skipTypeCheck,
-                                 String audience) {
+                                 String audience,
+                                 boolean enableBouncyCastleProvider,
+                                 int bouncyCastleProviderPosition) {
 
         if (keysEndpointUri == null) {
             throw new IllegalArgumentException("keysEndpointUri == null");
@@ -97,6 +107,20 @@ public JWTSignatureValidator(String keysEndpointUri,
         this.skipTypeCheck = skipTypeCheck;
         this.audience = audience;
 
+        if (enableBouncyCastleProvider && !bouncyInstalled.getAndSet(true)) {
+            int installedPosition = Security.insertProviderAt(new BouncyCastleProvider(), bouncyCastleProviderPosition);
+            log.info("BouncyCastle security provider installed at position: " + installedPosition);
+
+            if (log.isDebugEnabled()) {
+                StringBuilder sb = new StringBuilder("Installed security providers:\n");
+                for (Provider p: Security.getProviders()) {
+                    sb.append("  - " + p.toString() + "  [" + p.getClass().getName() + "]\n");
+                    sb.append("   " + p.getInfo() + "\n");
+                }
+                log.debug(sb.toString());
+            }
+        }
+
         fetchKeys();
 
         // set up periodic timer to update keys from server every refreshSeconds;
@@ -111,7 +135,9 @@ public JWTSignatureValidator(String keysEndpointUri,
                     + "\n    validIssuerUri: " + validIssuerUri
                     + "\n    certsRefreshSeconds: " + refreshSeconds
                     + "\n    certsExpirySeconds: " + expirySeconds
-                    + "\n    skipTypeCheck: " + skipTypeCheck);
+                    + "\n    skipTypeCheck: " + skipTypeCheck
+                    + "\n    enableBouncyCastleProvider: " + enableBouncyCastleProvider
+                    + "\n    bouncyCastleProviderPosition: " + bouncyCastleProviderPosition);
         }
     }
 
@@ -164,7 +190,6 @@ public TokenInfo validate(String token) {
             throw new TokenValidationException("Token signature validation failed: " + token, e)
                     .status(Status.INVALID_TOKEN);
         }
-        tokenVerifier.publicKey(getPublicKey(kid));
 
         if (audience != null) {
             tokenVerifier.audience(audience);
@@ -173,6 +198,20 @@ public TokenInfo validate(String token) {
         AccessToken t;
 
         try {
+            KeyWrapper keywrap = new KeyWrapper();
+            PublicKey pub = getPublicKey(kid);
+            keywrap.setPublicKey(pub);
+            keywrap.setAlgorithm(tokenVerifier.getHeader().getAlgorithm().name());
+            keywrap.setKid(kid);
+
+            log.debug("Signature algorithm used: [" + pub.getAlgorithm() + "]");
+            AsymmetricSignatureVerifierContext ctx = isAlgorithmEC(pub.getAlgorithm()) ?
+                    new ECDSASignatureVerifierContext(keywrap) :
+                    new AsymmetricSignatureVerifierContext(keywrap);
+            tokenVerifier.verifierContext(ctx);
+
+            log.debug("SignatureVerifierContext set to: " + ctx);
+
             tokenVerifier.verify();
             t = tokenVerifier.getToken();
 
@@ -193,6 +232,10 @@ public TokenInfo validate(String token) {
         return new TokenInfo(t, token);
     }
 
+    private static boolean isAlgorithmEC(String algorithm) {
+        return "EC".equals(algorithm) || "ECDSA".equals(algorithm);
+    }
+
 
     /**
      * Use daemon thread for refresh job
diff --git a/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/JaasServerOauthValidatorCallbackHandler.java b/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/JaasServerOauthValidatorCallbackHandler.java
index 542aae19..236b9bc6 100644
--- a/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/JaasServerOauthValidatorCallbackHandler.java
+++ b/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/JaasServerOauthValidatorCallbackHandler.java
@@ -62,12 +62,10 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
             throw new IllegalArgumentException("Exactly one jaasConfigEntry expected (size: " + jaasConfigEntries.size());
         }
 
-        for (AppConfigurationEntry e: jaasConfigEntries) {
-            Properties p = new Properties();
-            p.putAll(e.getOptions());
-            config = new ServerConfig(p);
-            break;
-        }
+        AppConfigurationEntry e = jaasConfigEntries.get(0);
+        Properties p = new Properties();
+        p.putAll(e.getOptions());
+        config = new ServerConfig(p);
 
         notJWT = config.getValueAsBoolean(Config.OAUTH_TOKENS_NOT_JWT, false);
 
@@ -79,6 +77,9 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         String jwksUri = config.getValue(ServerConfig.OAUTH_JWKS_ENDPOINT_URI);
 
+        boolean enableBouncy = config.getValueAsBoolean(ServerConfig.OAUTH_CRYPTO_PROVIDER_BOUNCYCASTLE, false);
+        int bouncyPosition = config.getValueAsInt(ServerConfig.OAUTH_CRYPTO_PROVIDER_BOUNCYCASTLE_POSITION, 0);
+
         if (jwksUri != null) {
             validator = new JWTSignatureValidator(
                     config.getValue(ServerConfig.OAUTH_JWKS_ENDPOINT_URI),
@@ -89,7 +90,9 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
                     config.getValueAsInt(ServerConfig.OAUTH_JWKS_EXPIRY_SECONDS, 360),
                     true,
                     config.getValueAsBoolean(ServerConfig.OAUTH_VALIDATION_SKIP_TYPE_CHECK, false),
-                    null
+                    null,
+                    enableBouncy,
+                    bouncyPosition
             );
         } else {
             validator = new OAuthIntrospectionValidator(
diff --git a/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/ServerConfig.java b/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/ServerConfig.java
index 38041bc9..c33e8944 100644
--- a/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/ServerConfig.java
+++ b/oauth-server/src/main/java/io/strimzi/kafka/oauth/server/ServerConfig.java
@@ -16,6 +16,8 @@ public class ServerConfig extends Config {
     public static final String OAUTH_VALID_ISSUER_URI = "oauth.valid.issuer.uri";
     public static final String OAUTH_INTROSPECTION_ENDPOINT_URI = "oauth.introspection.endpoint.uri";
     public static final String OAUTH_VALIDATION_SKIP_TYPE_CHECK = "oauth.validation.skip.type.check";
+    public static final String OAUTH_CRYPTO_PROVIDER_BOUNCYCASTLE = "oauth.crypto.provider.bouncycastle";
+    public static final String OAUTH_CRYPTO_PROVIDER_BOUNCYCASTLE_POSITION = "oauth.crypto.provider.bouncycastle.position";
 
     public ServerConfig() {
     }
diff --git a/pom.xml b/pom.xml
index 9e66c6cc..1fffcb7a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -72,6 +72,7 @@
         <junit.version>4.12</junit.version>
         <slf4j.version>1.7.26</slf4j.version>
         <keycloak.version>7.0.0</keycloak.version>
+        <bouncycastle.version>1.60</bouncycastle.version>
     </properties>
 
     <distributionManagement>
@@ -140,6 +141,11 @@
                 <artifactId>keycloak-common</artifactId>
                 <version>${keycloak.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.bouncycastle</groupId>
+                <artifactId>bcprov-jdk15on</artifactId>
+                <version>${bouncycastle.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.github.spotbugs</groupId>
                 <artifactId>spotbugs-annotations</artifactId>
diff --git a/testsuite/README.md b/testsuite/README.md
index 56b698ea..f4dab610 100644
--- a/testsuite/README.md
+++ b/testsuite/README.md
@@ -28,6 +28,11 @@ Also, when Kafka client connects to Kafka broker running inside docker image, th
 Running
 =======
 
+You may first need to perform the following cleanup of pre-existing containers / network definitions:
+
+    docker rm -f kafka zookeeper keycloak hydra
+    docker network rm $(docker network ls | grep test | awk '{print $1}')
+    
 To build and run the testsuite you need a running 'docker' daemon, then simply run:
 
     mvn clean install
diff --git a/testsuite/client-secret-jwt-keycloak-test/arquillian.xml b/testsuite/client-secret-jwt-keycloak-test/arquillian.xml
index 9e1b2fc5..3eb4ed0f 100644
--- a/testsuite/client-secret-jwt-keycloak-test/arquillian.xml
+++ b/testsuite/client-secret-jwt-keycloak-test/arquillian.xml
@@ -11,16 +11,24 @@
               await:
                 strategy: sleeping
                 sleepTime: 5 s
+
             kafka:
               await:
                 strategy: log
                 match: '[KafkaServer id=1] started'
                 timeout: 120
+              beforeStop:
+                - log:
+                    to: ${PWD}/../kafka.log
+                    stdout: true
+                    stderr: true
+
             keycloak:
               await:
                 strategy: log
                 match: 'regexp:.* Keycloak .* started in .*'
                 timeout: 120
+
         </property>
     </extension>
 </arquillian>
\ No newline at end of file
diff --git a/testsuite/client-secret-jwt-keycloak-test/docker-compose.yml b/testsuite/client-secret-jwt-keycloak-test/docker-compose.yml
index 069afe39..f25618df 100644
--- a/testsuite/client-secret-jwt-keycloak-test/docker-compose.yml
+++ b/testsuite/client-secret-jwt-keycloak-test/docker-compose.yml
@@ -24,7 +24,7 @@ services:
       - KEYCLOAK_PASSWORD=admin
       - KEYCLOAK_HTTPS_PORT=8443
       - PROXY_ADDRESS_FORWARDING=true
-      - KEYCLOAK_IMPORT=/opt/jboss/keycloak/realms/demo.json
+      - KEYCLOAK_IMPORT=/opt/jboss/keycloak/realms/demo-ec.json
 
       # Wait for up to 90 seconds for service to be up before ARQ Cube's wait_for_it gives up
       - TIMEOUT=90
@@ -59,18 +59,19 @@ services:
       # Authentication config
       - OAUTH_CLIENT_ID=kafka-broker
       - OAUTH_CLIENT_SECRET=kafka-broker-secret
-      - OAUTH_TOKEN_ENDPOINT_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo}/protocol/openid-connect/token
+      - OAUTH_TOKEN_ENDPOINT_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo-ec}/protocol/openid-connect/token
 
       # Validation config
-      - OAUTH_VALID_ISSUER_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo}
-      - OAUTH_JWKS_ENDPOINT_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo}/protocol/openid-connect/certs
+      - OAUTH_VALID_ISSUER_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo-ec}
+      - OAUTH_JWKS_ENDPOINT_URI=http://${KEYCLOAK_HOST:-keycloak}:8080/auth/realms/${REALM:-demo-ec}/protocol/openid-connect/certs
+      - OAUTH_CRYPTO_PROVIDER_BOUNCYCASTLE=true
 
       # username extraction from JWT token claim
       - OAUTH_USERNAME_CLAIM=preferred_username
 
       # For start.sh script to know where the keycloak is listening
       - KEYCLOAK_HOST=${KEYCLOAK_HOST:-keycloak}
-      - REALM=${REALM:-demo}
+      - REALM=${REALM:-demo-ec}
 
       # Wait for up to 90 seconds for service to be up before ARQ Cube's wait_for_it gives up
       - TIMEOUT=90
diff --git a/testsuite/client-secret-jwt-keycloak-test/src/test/java/io/strimzi/testsuite/oauth/KeycloakClientCredentialsWithJwtValidationTest.java b/testsuite/client-secret-jwt-keycloak-test/src/test/java/io/strimzi/testsuite/oauth/KeycloakClientCredentialsWithJwtValidationTest.java
index 461bae28..c8b90e42 100644
--- a/testsuite/client-secret-jwt-keycloak-test/src/test/java/io/strimzi/testsuite/oauth/KeycloakClientCredentialsWithJwtValidationTest.java
+++ b/testsuite/client-secret-jwt-keycloak-test/src/test/java/io/strimzi/testsuite/oauth/KeycloakClientCredentialsWithJwtValidationTest.java
@@ -30,11 +30,11 @@
 public class KeycloakClientCredentialsWithJwtValidationTest {
 
     private static final String HOST = "keycloak";
-    private static final String REALM = "demo";
+    private static final String REALM = "demo-ec";
 
     @Test
     public void doTest() throws Exception {
-        System.out.println("==== KeycloakClientCredentialsWithJwtValidationTest ====");
+        System.out.println("==== KeycloakClientCredentialsWithJwtValidationTest + test EC ====");
 
         Properties p = System.getProperties();
         for (Object key: p.keySet()) {
diff --git a/testsuite/docker/keycloak/realms/demo-ec.json b/testsuite/docker/keycloak/realms/demo-ec.json
new file mode 100644
index 00000000..adb34b7d
--- /dev/null
+++ b/testsuite/docker/keycloak/realms/demo-ec.json
@@ -0,0 +1,243 @@
+{
+  "realm": "demo-ec",
+  "accessTokenLifespan": 300,
+  "ssoSessionMaxLifespan": 32140800,
+  "ssoSessionIdleTimeout": 32140800,
+  "enabled": true,
+  "sslRequired": "external",
+  "defaultSignatureAlgorithm": "ES256",
+  "components": {
+    "org.keycloak.keys.KeyProvider": [
+      {
+        "name": "ES256",
+        "providerId": "ecdsa-generated",
+        "subComponents": {},
+        "config": {
+          "ecdsaEllipticCurveKey": [
+            "P-256"
+          ],
+          "active": [
+            "true"
+          ],
+          "priority": [
+            "101"
+          ],
+          "enabled": [
+            "true"
+          ]
+        }
+      }
+    ]
+  },
+  "users": [
+    {
+      "username": "alice",
+      "enabled": true,
+      "email": "alice@example.com",
+      "credentials": [
+        {
+          "type": "password",
+          "value": "alice-password"
+        }
+      ],
+      "realmRoles": [
+        "user"
+      ],
+      "clientRoles": {
+        "kafka": [
+          "kafka-topic:superapp_*:owner"
+        ]
+      }
+    },
+    {
+      "username": "admin",
+      "enabled": true,
+      "email": "admin@example.com",
+      "credentials": [
+        {
+          "type": "password",
+          "value": "admin-password"
+        }
+      ],
+      "realmRoles": [
+        "admin"
+      ],
+      "clientRoles": {
+        "realm-management": [
+          "realm-admin"
+        ],
+        "kafka": [
+          "kafka-admin"
+        ]
+      }
+    },
+    {
+      "username": "service-account-kafka-broker",
+      "enabled": true,
+      "email": "service-account-kafka-broker@placeholder.org",
+      "serviceAccountClientId": "kafka-broker",
+      "clientRoles": {
+        "kafka" : ["kafka-admin"]
+      }
+    },
+    {
+      "username": "service-account-kafka-producer-client",
+      "enabled": true,
+      "email": "service-account-kafka-producer-client@placeholder.org",
+      "serviceAccountClientId": "kafka-producer-client"
+    },
+    {
+      "username": "service-account-kafka-consumer-client",
+      "enabled": true,
+      "email": "service-account-kafka-consumer-client@placeholder.org",
+      "serviceAccountClientId": "kafka-consumer-client",
+      "clientRoles": {
+        "kafka" : ["kafka-topic:superapp_*:consumer"]
+      }
+    }
+  ],
+  "roles": {
+    "realm": [
+      {
+        "name": "user",
+        "description": "User privileges"
+      },
+      {
+        "name": "admin",
+        "description": "Administrator privileges"
+      }
+    ],
+    "client": {
+      "kafka": [
+        {
+          "name": "kafka-admin",
+          "description": "Kafka administrator - can perform any action on any Kafka resource",
+          "clientRole": true
+        },
+        {
+          "name": "kafka-topic:superapp_*:owner",
+          "description": "Owner of topics that begin with 'superapp_' prefix. Can perform any operation on these topics.",
+          "clientRole": true
+        },
+        {
+          "name": "kafka-topic:superapp_*:consumer",
+          "description": "Consumer of topics that begin with 'superapp_' prefix. Can perform READ, and DESCRIBE on these topics.",
+          "clientRole": true
+        }
+      ]
+    }
+  },
+  "scopeMappings": [
+    {
+      "client": "kafka-broker",
+      "roles": [
+        "offline_access"
+      ]
+    },
+    {
+      "client": "kafka-producer-client",
+      "roles": [
+        "offline_access"
+      ]
+    },
+    {
+      "client": "kafka-consumer-client",
+      "roles": [
+        "offline_access"
+      ]
+    },
+    {
+      "clientScope": "offline_access",
+      "roles": [
+        "offline_access"
+      ]
+    }
+  ],
+  "clientScopeMappings": {
+    "kafka": [
+      {
+        "client": "kafka-broker",
+        "roles": [
+          "kafka-admin"
+        ]
+      },
+      {
+        "client": "kafka-consumer-client",
+        "roles": [
+          "kafka-topic:superapp_*:consumer"
+        ]
+      },
+      {
+        "client": "kafka-producer-client",
+        "roles": [
+          "kafka-topic:superapp_*:owner"
+        ]
+      }
+    ]
+  },
+  "clients": [
+    {
+      "clientId": "kafka",
+      "enabled": true,
+      "publicClient": true,
+      "bearerOnly": false,
+      "standardFlowEnabled": false,
+      "implicitFlowEnabled": false,
+      "directAccessGrantsEnabled": false,
+      "serviceAccountsEnabled": false,
+      "consentRequired" : false,
+      "fullScopeAllowed" : false
+    },
+    {
+      "clientId": "kafka-broker",
+      "enabled": true,
+      "clientAuthenticatorType": "client-secret",
+      "secret": "kafka-broker-secret",
+      "publicClient": false,
+      "bearerOnly": false,
+      "standardFlowEnabled": false,
+      "implicitFlowEnabled": false,
+      "directAccessGrantsEnabled": true,
+      "serviceAccountsEnabled": true,
+      "consentRequired" : false,
+      "fullScopeAllowed" : false,
+      "attributes": {
+        "access.token.lifespan": "32140800"
+      }
+    },
+    {
+      "clientId": "kafka-producer-client",
+      "enabled": true,
+      "clientAuthenticatorType": "client-secret",
+      "secret": "kafka-producer-client-secret",
+      "publicClient": false,
+      "bearerOnly": false,
+      "standardFlowEnabled": false,
+      "implicitFlowEnabled": false,
+      "directAccessGrantsEnabled": true,
+      "serviceAccountsEnabled": true,
+      "consentRequired" : false,
+      "fullScopeAllowed" : false,
+      "attributes": {
+        "access.token.lifespan": "36000"
+      }
+    },
+    {
+      "clientId": "kafka-consumer-client",
+      "enabled": true,
+      "clientAuthenticatorType": "client-secret",
+      "secret": "kafka-consumer-client-secret",
+      "publicClient": false,
+      "bearerOnly": false,
+      "standardFlowEnabled": false,
+      "implicitFlowEnabled": false,
+      "directAccessGrantsEnabled": true,
+      "serviceAccountsEnabled": true,
+      "consentRequired" : false,
+      "fullScopeAllowed" : false,
+      "attributes": {
+        "access.token.lifespan": "32140800"
+      }
+    }
+  ]
+}
\ No newline at end of file