diff --git a/README.md b/README.md
index 06e5a2bd..7e3942bd 100644
--- a/README.md
+++ b/README.md
@@ -945,6 +945,10 @@ together with one of authentication options below.
 
 When client starts to establish the connection with the Kafka Broker it will first obtain an access token from the configured Token Endpoint, authenticating with the configured client ID and configured authentication option using client_credentials grant type.
 
+If the OAuth2 server is using an alternative to the "grant_type=client_credentials" string, such as "grant_type=kubernetes", that is achieved by specifying the following:
+- `oauth.client.credentials.grant.type` (e.g.: "kubernetes")
+
+
 ##### Option 1: Using a Client Secret 
 
 Specify the client secret.
diff --git a/oauth-client/src/main/java/io/strimzi/kafka/oauth/client/JaasClientOauthLoginCallbackHandler.java b/oauth-client/src/main/java/io/strimzi/kafka/oauth/client/JaasClientOauthLoginCallbackHandler.java
index 53b9eb1f..919950a8 100644
--- a/oauth-client/src/main/java/io/strimzi/kafka/oauth/client/JaasClientOauthLoginCallbackHandler.java
+++ b/oauth-client/src/main/java/io/strimzi/kafka/oauth/client/JaasClientOauthLoginCallbackHandler.java
@@ -73,6 +73,7 @@ public class JaasClientOauthLoginCallbackHandler implements AuthenticateCallback
     private String scope;
     private String audience;
     private URI tokenEndpoint;
+    private String grantType;
 
     private boolean isJwt;
     private int maxTokenExpirySeconds;
@@ -152,6 +153,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         scope = config.getValue(Config.OAUTH_SCOPE);
         audience = config.getValue(Config.OAUTH_AUDIENCE);
+        grantType = config.getValue(Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
         socketFactory = ConfigUtil.createSSLFactory(config);
         hostnameVerifier = ConfigUtil.createHostnameVerifier(config);
         connectTimeout = getConnectTimeout(config);
@@ -208,6 +210,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
                     + "\n    password: " + mask(password)
                     + "\n    scope: " + scope
                     + "\n    audience: " + audience
+                    + "\n    grantType: " + grantType
                     + "\n    isJwt: " + isJwt
                     + "\n    maxTokenExpirySeconds: " + maxTokenExpirySeconds
                     + "\n    principalExtractor: " + principalExtractor
@@ -397,9 +400,9 @@ private void handleCallback(OAuthBearerTokenCallback callback) throws IOExceptio
             } else if (username != null) {
                 result = loginWithPassword(tokenEndpoint, socketFactory, hostnameVerifier, username, password, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
             } else if (clientSecret != null) {
-                result = loginWithClientSecret(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
+                result = loginWithClientSecret(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientSecret, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader, grantType);
             } else if (clientAssertionProvider != null) {
-                result = loginWithClientAssertion(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientAssertionProvider.token(), clientAssertionType, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader);
+                result = loginWithClientAssertion(tokenEndpoint, socketFactory, hostnameVerifier, clientId, clientAssertionProvider.token(), clientAssertionType, isJwt, principalExtractor, scope, audience, connectTimeout, readTimeout, authenticatorMetrics, retries, retryPauseMillis, includeAcceptHeader, grantType);
             } else {
                 throw new IllegalStateException("Invalid oauth client configuration - no credentials");
             }
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 3fa27cd3..92b25b34 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
@@ -23,7 +23,14 @@ public class Config {
     /** The name of 'oauth.client.secret' config option  */
     public static final String OAUTH_CLIENT_SECRET = "oauth.client.secret";
 
-    /** The name of 'oauth.scope' config option  */
+    /** The name of 'oauth.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE = "oauth.client.credentials.grant.type";
+
+    /** The fallback for 'oauth.client.credentials.grant.type.string' config option  */
+    public static final String OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK = "client_credentials";
+
+
+  /** The name of 'oauth.scope' config option  */
     public static final String OAUTH_SCOPE = "oauth.scope";
 
     /** The name of 'oauth.audience' config option  */
diff --git a/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java b/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java
index 7024f76a..88c17abb 100644
--- a/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java
+++ b/oauth-common/src/main/java/io/strimzi/kafka/oauth/common/OAuthAuthenticator.java
@@ -73,6 +73,7 @@ public static TokenInfo loginWithAccessToken(String token, boolean isJwt, Princi
      * @param principalExtractor A PrincipalExtractor to use to determine the principal (user id)
      * @param scope A scope to request when authenticating
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -80,10 +81,11 @@ public static TokenInfo loginWithAccessToken(String token, boolean isJwt, Princi
     public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFactory socketFactory,
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
-                                                  PrincipalExtractor principalExtractor, String scope, boolean includeAcceptHeader) throws IOException {
+                                                  PrincipalExtractor principalExtractor, String scope, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
 
         return loginWithClientSecret(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientSecret, isJwt, principalExtractor, scope, null, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader);
+                clientId, clientSecret, isJwt, principalExtractor, scope, null, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader, grantType);
     }
 
     /**
@@ -100,6 +102,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param scope A scope to request when authenticating
      * @param audience An 'audience' attribute to set on the request when authenticating
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -107,10 +110,11 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
     public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFactory socketFactory,
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
-                                                  PrincipalExtractor principalExtractor, String scope, String audience, boolean includeAcceptHeader) throws IOException {
+                                                  PrincipalExtractor principalExtractor, String scope, String audience, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
 
         return loginWithClientSecret(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientSecret, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader);
+                clientId, clientSecret, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, includeAcceptHeader, grantType);
     }
 
     /**
@@ -132,6 +136,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param retries A maximum number of retries if the request fails due to network, or unexpected response status
      * @param retryPauseMillis A pause between consecutive requests
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -141,7 +146,8 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
                                                   HostnameVerifier hostnameVerifier,
                                                   String clientId, String clientSecret, boolean isJwt,
                                                   PrincipalExtractor principalExtractor, String scope, String audience,
-                                                  int connectTimeout, int readTimeout, MetricsHandler metrics, int retries, long retryPauseMillis, boolean includeAcceptHeader) throws IOException {
+                                                  int connectTimeout, int readTimeout, MetricsHandler metrics, int retries, long retryPauseMillis, boolean includeAcceptHeader,
+                                                  String grantType) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug("loginWithClientSecret() - tokenEndpointUrl: {}, clientId: {}, clientSecret: {}, scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
                     tokenEndpointUrl, clientId, mask(clientSecret), scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
@@ -156,7 +162,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
 
         String authorization = "Basic " + base64encode(clientId + ':' + clientSecret);
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials");
+        StringBuilder body = new StringBuilder("grant_type=" + grantType);
         if (scope != null) {
             body.append("&scope=").append(urlencode(scope));
         }
@@ -182,6 +188,7 @@ public static TokenInfo loginWithClientSecret(URI tokenEndpointUrl, SSLSocketFac
      * @param principalExtractor A PrincipalExtractor to use to determine the principal (user id)
      * @param scope A scope to request when authenticating
      * @param audience An 'audience' attribute to set on the request when authenticating
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -196,10 +203,11 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
                                                      boolean isJwt,
                                                      PrincipalExtractor principalExtractor,
                                                      String scope,
-                                                     String audience) throws IOException {
+                                                     String audience,
+                                                     String grantType) throws IOException {
 
         return loginWithClientAssertion(tokenEndpointUrl, socketFactory, hostnameVerifier,
-                clientId, clientAssertion, clientAssertionType, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, true);
+                clientId, clientAssertion, clientAssertionType, isJwt, principalExtractor, scope, audience, HttpUtil.DEFAULT_CONNECT_TIMEOUT, HttpUtil.DEFAULT_READ_TIMEOUT, null, 0, 0, true, grantType);
     }
 
     /**
@@ -222,6 +230,7 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
      * @param retries A maximum number of retries if the request fails due to network, or unexpected response status
      * @param retryPauseMillis A pause between consecutive requests
      * @param includeAcceptHeader Should we skip sending the Accept header when making outbound http requests
+     * @param grantType The grant type to be used, typically "client_credentials"
      * @return A TokenInfo with access token and information extracted from it
      * @throws IOException If the request to the authorization server has failed
      * @throws IllegalStateException If the response from the authorization server could not be handled
@@ -242,7 +251,8 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
                                                      MetricsHandler metrics,
                                                      int retries,
                                                      long retryPauseMillis,
-                                                     boolean includeAcceptHeader) throws IOException {
+                                                     boolean includeAcceptHeader,
+                                                     String grantType) throws IOException {
         if (log.isDebugEnabled()) {
             log.debug("loginWithClientAssertion() - tokenEndpointUrl: {}, clientId: {}, clientAssertion: {}, clientAssertionType: {}, scope: {}, audience: {}, connectTimeout: {}, readTimeout: {}, retries: {}, retryPauseMillis: {}",
                     tokenEndpointUrl, clientId, mask(clientAssertion), clientAssertionType, scope, audience, connectTimeout, readTimeout, retries, retryPauseMillis);
@@ -252,7 +262,7 @@ public static TokenInfo loginWithClientAssertion(URI tokenEndpointUrl,
             throw new IllegalArgumentException("No clientId specified");
         }
 
-        StringBuilder body = new StringBuilder("grant_type=client_credentials")
+        StringBuilder body = new StringBuilder("grant_type=" + grantType)
                 .append("&client_id=").append(urlencode(clientId))
                 .append("&client_assertion=").append(urlencode(clientAssertion))
                 .append("&client_assertion_type=").append(urlencode(clientAssertionType));
diff --git a/oauth-server-plain/src/main/java/io/strimzi/kafka/oauth/server/plain/JaasServerOauthOverPlainValidatorCallbackHandler.java b/oauth-server-plain/src/main/java/io/strimzi/kafka/oauth/server/plain/JaasServerOauthOverPlainValidatorCallbackHandler.java
index 614cacc2..8fab1ab0 100644
--- a/oauth-server-plain/src/main/java/io/strimzi/kafka/oauth/server/plain/JaasServerOauthOverPlainValidatorCallbackHandler.java
+++ b/oauth-server-plain/src/main/java/io/strimzi/kafka/oauth/server/plain/JaasServerOauthOverPlainValidatorCallbackHandler.java
@@ -120,6 +120,7 @@ public class JaasServerOauthOverPlainValidatorCallbackHandler extends JaasServer
     private URI tokenEndpointUri;
     private String scope;
     private String audience;
+    private String grantType;
 
     private OAuthMetrics metrics;
     private boolean enableMetrics;
@@ -148,6 +149,7 @@ public void configure(Map<String, ?> configs, String saslMechanism, List<AppConf
 
         scope = config.getValue(ServerConfig.OAUTH_SCOPE);
         audience = config.getValue(ServerConfig.OAUTH_AUDIENCE);
+        grantType = config.getValue(Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         super.delegatedConfigure(configs, "PLAIN", jaasConfigEntries);
 
@@ -247,7 +249,7 @@ private void authenticate(String username, String password) throws UnsupportedCa
                 checkUsernameMatch = true;
             } else if (tokenEndpointUri != null) {
                 accessToken = OAuthAuthenticator.loginWithClientSecret(tokenEndpointUri, getSocketFactory(), getVerifier(),
-                        username, password, isJwt(), getPrincipalExtractor(), scope, audience, getConnectTimeout(), getReadTimeout(), authMetrics, getRetries(), getRetryPauseMillis(), includeAcceptHeader())
+                        username, password, isJwt(), getPrincipalExtractor(), scope, audience, getConnectTimeout(), getReadTimeout(), authMetrics, getRetries(), getRetryPauseMillis(), includeAcceptHeader(), grantType)
                         .token();
             } else {
                 throw new ValidationException("Empty password where access token was expected");
diff --git a/testsuite/hydra-test/src/test/java/io/strimzi/testsuite/oauth/HydraAuthenticationTest.java b/testsuite/hydra-test/src/test/java/io/strimzi/testsuite/oauth/HydraAuthenticationTest.java
index ec862755..a59d6b12 100644
--- a/testsuite/hydra-test/src/test/java/io/strimzi/testsuite/oauth/HydraAuthenticationTest.java
+++ b/testsuite/hydra-test/src/test/java/io/strimzi/testsuite/oauth/HydraAuthenticationTest.java
@@ -42,6 +42,7 @@
 import java.util.Map;
 import java.util.Properties;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.common.TestUtil.logStart;
 
 /**
@@ -135,7 +136,7 @@ public void opaqueAccessTokenWithIntrospectValidationTest(String title) throws E
         // first, request access token using client id and secret
         TokenInfo info = OAuthAuthenticator.loginWithClientSecret(URI.create(tokenEndpointUri),
                 ConfigUtil.createSSLFactory(new ClientConfig()),
-                null, clientId, clientSecret, true, null, null, true);
+                null, clientId, clientSecret, true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         oauthConfig.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, tokenEndpointUri);
diff --git a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java
index 3a7e4356..8f9b1e00 100644
--- a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java
+++ b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/BasicTests.java
@@ -28,6 +28,7 @@
 import java.util.Properties;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.kafka.oauth.common.TokenIntrospection.introspectAccessToken;
 import static io.strimzi.testsuite.oauth.auth.Common.buildConsumerConfigOAuthBearer;
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigOAuthBearer;
@@ -253,7 +254,8 @@ void accessTokenWithIntrospection() throws Exception {
         final String clientSecret = "kafka-producer-client-secret";
 
         // First, request access token using client id and secret
-        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true);
+        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true,
+                                               OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         oauthConfig.put(ClientConfig.OAUTH_ACCESS_TOKEN, info.token());
diff --git a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/JwtManipulationTests.java b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/JwtManipulationTests.java
index fb132668..cd0fff8b 100644
--- a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/JwtManipulationTests.java
+++ b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/JwtManipulationTests.java
@@ -40,6 +40,7 @@
 
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigOAuthBearer;
 import static io.strimzi.testsuite.oauth.auth.Common.loginWithUsernamePassword;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 
 public class JwtManipulationTests {
 
@@ -224,7 +225,8 @@ private String getOriginalToken() throws IOException {
 
         // first, request access token using client id and secret
         TokenInfo info = OAuthAuthenticator.loginWithClientSecret(URI.create(tokenEndpointUri), null, null,
-                "kafka-producer-client", "kafka-producer-client-secret", true, null, null, true);
+                "kafka-producer-client", "kafka-producer-client-secret", true, null, null, true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         return info.token();
     }
diff --git a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/OAuthOverPlainTests.java b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/OAuthOverPlainTests.java
index 0d04ae63..35548ef1 100644
--- a/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/OAuthOverPlainTests.java
+++ b/testsuite/keycloak-auth-tests/src/test/java/io/strimzi/testsuite/oauth/auth/OAuthOverPlainTests.java
@@ -27,6 +27,7 @@
 import java.util.concurrent.ExecutionException;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.auth.Common.buildConsumerConfigPlain;
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigPlain;
 import static io.strimzi.testsuite.oauth.auth.Common.poll;
@@ -64,7 +65,7 @@ static void accessTokenOverPlainWithClientCredentialsDisabled() throws Exception
 
         // first, request access token using client id and secret
         TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null,
-                "team-a-client", "team-a-client-secret", true, null, null, true);
+                "team-a-client", "team-a-client-secret", true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> plainConfig = new HashMap<>();
         plainConfig.put("username", "service-account-team-a-client");
@@ -218,7 +219,7 @@ static void accessTokenOverPlainWithIntrospection() throws Exception {
 
         // first, request access token using client id and secret
         TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null,
-                "team-a-client", "team-a-client-secret", true, null, null, true);
+                "team-a-client", "team-a-client-secret", true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> plainConfig = new HashMap<>();
         plainConfig.put("username", "service-account-team-a-client");
diff --git a/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/Common.java b/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/Common.java
index a61176ba..87723c1b 100644
--- a/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/Common.java
+++ b/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/Common.java
@@ -44,6 +44,7 @@
 import java.util.concurrent.ExecutionException;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.urlencode;
 
 @SuppressFBWarnings({"THROWS_METHOD_THROWS_CLAUSE_BASIC_EXCEPTION", "THROWS_METHOD_THROWS_RUNTIMEEXCEPTION", "RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"})
@@ -93,9 +94,9 @@ static void produceToTopic(String topic, Properties config) throws Exception {
 
     void authenticateAllActors() throws IOException {
         tokens.put(TEAM_A_CLIENT, loginWithClientSecret(URI.create(TOKEN_ENDPOINT_URI), null, null,
-                TEAM_A_CLIENT, TEAM_A_CLIENT + "-secret", true, null, null, true).token());
+                TEAM_A_CLIENT, TEAM_A_CLIENT + "-secret", true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK).token());
         tokens.put(TEAM_B_CLIENT, loginWithClientSecret(URI.create(TOKEN_ENDPOINT_URI), null, null,
-                TEAM_B_CLIENT, TEAM_B_CLIENT + "-secret", true, null, null, true).token());
+                TEAM_B_CLIENT, TEAM_B_CLIENT + "-secret", true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK).token());
         tokens.put(BOB, loginWithUsernamePassword(URI.create(TOKEN_ENDPOINT_URI),
                 BOB, BOB + "-password", "kafka-cli"));
         tokens.put(ZERO, loginWithUsernamePassword(URI.create(TOKEN_ENDPOINT_URI),
diff --git a/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/FloodTest.java b/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/FloodTest.java
index efb7b6f6..7676a25e 100644
--- a/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/FloodTest.java
+++ b/testsuite/keycloak-authz-kraft-tests/src/test/java/io/strimzi/testsuite/oauth/authz/FloodTest.java
@@ -26,6 +26,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 
 @SuppressFBWarnings({"THROWS_METHOD_THROWS_RUNTIMEEXCEPTION", "THROWS_METHOD_THROWS_CLAUSE_THROWABLE"})
 public class FloodTest extends Common {
@@ -159,7 +160,7 @@ private void obtainAndStoreToken(String producerPrefix, HashMap<String, String>
         String secret = clientId + "-secret";
 
         tokens.put(clientId, loginWithClientSecret(URI.create(TOKEN_ENDPOINT_URI), null, null,
-                clientId, secret, true, null, null, true).token());
+                clientId, secret, true, null, null, true, OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK).token());
     }
 
 
diff --git a/testsuite/keycloak-errors-tests/src/test/java/io/strimzi/testsuite/oauth/auth/ErrorReportingTests.java b/testsuite/keycloak-errors-tests/src/test/java/io/strimzi/testsuite/oauth/auth/ErrorReportingTests.java
index bb20ca78..c60d9af2 100644
--- a/testsuite/keycloak-errors-tests/src/test/java/io/strimzi/testsuite/oauth/auth/ErrorReportingTests.java
+++ b/testsuite/keycloak-errors-tests/src/test/java/io/strimzi/testsuite/oauth/auth/ErrorReportingTests.java
@@ -21,6 +21,7 @@
 import java.util.concurrent.ExecutionException;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.loginWithClientSecret;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigOAuthBearer;
 import static io.strimzi.testsuite.oauth.auth.Common.buildProducerConfigPlain;
 import static io.strimzi.testsuite.oauth.common.TestUtil.assertTrueExtra;
@@ -172,7 +173,8 @@ private void forgedJwtSig() throws Exception {
         final String clientSecret = "kafka-producer-client-secret";
 
         // first, request access token using client id and secret
-        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true);
+        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true,
+                                               OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         String tokenWithBrokenSig = info.token().substring(0, info.token().length() - 6) + "ffffff";
@@ -213,7 +215,8 @@ private void forgedJwtSigIntrospect() throws Exception {
         final String clientSecret = "kafka-producer-client-secret";
 
         // first, request access token using client id and secret
-        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true);
+        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true,
+                                               OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         String tokenWithBrokenSig = info.token().substring(0, info.token().length() - 6) + "ffffff";
@@ -254,7 +257,8 @@ private void expiredJwtToken() throws Exception {
         final String clientSecret = "kafka-producer-client-secret";
 
         // first, request access token using client id and secret
-        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true);
+        TokenInfo info = loginWithClientSecret(URI.create(tokenEndpointUri), null, null, clientId, clientSecret, true, null, null, true,
+                                               OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         Map<String, String> oauthConfig = new HashMap<>();
         oauthConfig.put(ClientConfig.OAUTH_ACCESS_TOKEN, info.token());
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/AuthorizationEndpointsTest.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/AuthorizationEndpointsTest.java
index 8a17979b..91ec7d88 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/AuthorizationEndpointsTest.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/AuthorizationEndpointsTest.java
@@ -27,6 +27,7 @@
 import java.util.Map;
 
 import static io.strimzi.kafka.oauth.common.IOUtil.randomHexString;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.common.TestUtil.getRootCause;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.changeAuthServerMode;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.createOAuthClient;
@@ -67,7 +68,8 @@ public void doTest() throws Exception {
                 true,
                 null,
                 null,
-                true);
+                true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         OAuthBearerValidatorCallback[] oauthCallbacks = {new OAuthBearerValidatorCallback(tokenInfo.token())};
 
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/ClientAssertionAuthTest.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/ClientAssertionAuthTest.java
index f0c5ebb5..1cda7809 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/ClientAssertionAuthTest.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/ClientAssertionAuthTest.java
@@ -18,6 +18,7 @@
 import javax.net.ssl.SSLSocketFactory;
 import java.net.URI;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.WWW_FORM_CONTENT_TYPE;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.changeAuthServerMode;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.createOAuthClient;
@@ -59,7 +60,8 @@ public void doTest() throws Exception {
                     true,
                     null,
                     null,
-                    null);
+                    null,
+                    OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
             Assert.fail("Should have failed with 401");
         } catch (Exception e) {
@@ -79,7 +81,8 @@ public void doTest() throws Exception {
                 true,
                 null,
                 null,
-                null);
+                null,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         String token = tokenInfo.token();
         Assert.assertNotNull(token);
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/Common.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/Common.java
index 8b794ff5..5399c8d6 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/Common.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/Common.java
@@ -32,6 +32,7 @@
 import java.util.Properties;
 
 import static io.strimzi.kafka.oauth.common.OAuthAuthenticator.base64encode;
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 
 public class Common {
 
@@ -112,7 +113,8 @@ static String loginWithClientSecret(String tokenEndpoint, String clientId, Strin
                 new PrincipalExtractor(),
                 "all",
                 null,
-                true);
+                true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         return tokenInfo.token();
     }
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JWKSKeyUseTest.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JWKSKeyUseTest.java
index 0f46e738..e708e06a 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JWKSKeyUseTest.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JWKSKeyUseTest.java
@@ -20,6 +20,7 @@
 import java.net.URI;
 import java.util.Collections;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.changeAuthServerMode;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.createOAuthClient;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.getProjectRoot;
@@ -54,7 +55,8 @@ public void doTest() throws Exception {
                 null,
                 null,
                 null,
-                true);
+                true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         TokenIntrospection.debugLogJWT(log, tokenInfo.token());
 
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JaasClientConfigTest.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JaasClientConfigTest.java
index b2c3a8ba..2b594ea6 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JaasClientConfigTest.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/JaasClientConfigTest.java
@@ -78,6 +78,8 @@ public void doTest() throws Exception {
         testRefreshTokenLocation();
 
         testClientAssertionLocation();
+
+        testInvalidGrantType();
     }
 
     private void testAllConfigOptions() throws IOException {
@@ -146,6 +148,7 @@ private void testAllConfigOptions() throws IOException {
             "password", "p\\*\\*",
             "scope", "scope",
             "audience", "audience",
+            "grantType", ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK,
             "isJwt", "false",
             "usernameClaim", "username-claim",
             "fallbackUsernameClaim", "fallback-username-claim",
@@ -569,6 +572,45 @@ private void testClientAssertionLocation() throws Exception {
         }
     }
 
+    private void testInvalidGrantType() throws Exception {
+        String testClient = "testclient";
+        String testSecret = "testsecret";
+
+        changeAuthServerMode("jwks", "mode_200");
+        changeAuthServerMode("token", "mode_200");
+        createOAuthClient(testClient, testSecret);
+
+        Map<String, String> oauthConfig = new HashMap<>();
+        oauthConfig.put(ClientConfig.OAUTH_TOKEN_ENDPOINT_URI, TOKEN_ENDPOINT_URI);
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_ID, testClient);
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_SECRET, testSecret);
+        oauthConfig.put(ClientConfig.OAUTH_SSL_TRUSTSTORE_LOCATION, "../docker/target/kafka/certs/ca-truststore.p12");
+        oauthConfig.put(ClientConfig.OAUTH_SSL_TRUSTSTORE_PASSWORD, "changeit");
+
+        // Confirm fails with invalid grant type
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, "dummy-grant-type");
+
+        try {
+            initJaasWithRetry(oauthConfig);
+            Assert.fail("Should have failed");
+
+        } catch (KafkaException e) {
+            assertLoginException(e);
+        }
+
+        // Confirm succeeds with valid grant type
+        oauthConfig.put(ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE, ClientConfig.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
+
+        LogLineReader logReader = new LogLineReader(Common.LOG_PATH);
+        logReader.readNext();
+        initJaasWithRetry(oauthConfig);
+        List<String> lines = logReader.readNext();
+        boolean found = checkLogForRegex(lines, "Login succeeded");
+        Assert.assertTrue("Login succeeded", found);
+
+    }
+
+
     /**
      * If signing keys have not yet been loaded by kafka broker,
      * keep trying for up to 10 attempts with 2 second pause.
diff --git a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/PasswordAuthAndPrincipalExtractionTest.java b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/PasswordAuthAndPrincipalExtractionTest.java
index 57981fc7..cceea519 100644
--- a/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/PasswordAuthAndPrincipalExtractionTest.java
+++ b/testsuite/mockoauth-tests/src/test/java/io/strimzi/testsuite/oauth/mockoauth/PasswordAuthAndPrincipalExtractionTest.java
@@ -23,6 +23,7 @@
 import java.net.URI;
 import java.text.ParseException;
 
+import static io.strimzi.kafka.oauth.common.Config.OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.WWW_FORM_CONTENT_TYPE;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.changeAuthServerMode;
 import static io.strimzi.testsuite.oauth.mockoauth.Common.createOAuthClient;
@@ -132,7 +133,8 @@ public void doTest() throws Exception {
                 null,
                 null,
                 null,
-                true);
+                true,
+                OAUTH_CLIENT_CREDENTIALS_GRANT_TYPE_FALLBACK);
 
         token = tokenInfo.token();
         Assert.assertNotNull(token);