OAuth in Kafka with Keycloak

[sahil-sharma]

Hello everyone,

I am trying to unwrap my head around setting up OAuth in Kafka with Keycloak. Tried out setting up as mentioned in the documentation along with examples in GH.

I set-up Keycloak in my local kind cluster and configured it with kafka-authz.json (in GH example linked above).
I also read kafka-oauth documentation for the same to confirm OAuth workings.

I am facing some issues which are not clear to me as I am following as suggested.

Doubt-1:
As we are delegating the ACLs to keycloak so kafka ACLs would not interfere unless configured.
In Strimzi Operator examples it is OFF as here and mentioned in the documentation here.

strimzi.authorization.delegate.to.kafka.acl: "false"

while kafka-oauth code files has enabled it. See here

To rule this out I did exactly as mentioned in the official documentation:

strimzi.authorization.delegate.to.kafka.acl: "false"

and, removed this from my config:

 delegateToKafkaAcls: true

And, I faced an error in my kafka broker pod logs:

2025-12-29 11:50:01 ERROR [data-plane-kafka-request-handler-5] KafkaApis:76 - [KafkaApi-0] Unexpected error handling request RequestHeader(apiKey=INIT_PRODUCER_ID, apiVersion=5, clientId=console-producer, correlationId=2, headerVersion=2) -- InitProducerIdRequestData(transactionalId=null, transactionTimeoutMs=2147483647, producerId=-1, producerEpoch=-1) with context RequestContext(header=RequestHeader(apiKey=INIT_PRODUCER_ID, apiVersion=5, clientId=console-producer, correlationId=2, headerVersion=2), connectionId='10.244.2.11:9092-127.0.0.6:34185-3-1', clientAddress=/127.0.0.6, principal=OAuthKafkaPrincipal(User:service-account-team-a-client, groups: null, session: 909467807, token: eyJh**BDtw), listenerName=ListenerName(PLAIN-9092), securityProtocol=SASL_PLAINTEXT, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=4.0.0), fromPrivilegedListener=false, principalSerde=Optional[io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder@68849c7f])
java.lang.UnsupportedOperationException: Simple ACL delegation not enabled
	at io.strimzi.kafka.oauth.server.authorizer.KeycloakRBACAuthorizer.acls(KeycloakRBACAuthorizer.java:599) ~[kafka-oauth-keycloak-authorizer-0.17.0.jar:?]
	at org.apache.kafka.server.authorizer.Authorizer.authorizeByResourceType(Authorizer.java:217) ~[kafka-clients-4.0.0.jar:?]
	at io.strimzi.kafka.oauth.server.authorizer.KeycloakAuthorizer.authorizeByResourceType(KeycloakAuthorizer.java:218) ~[kafka-oauth-keycloak-authorizer-0.17.0.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorizeByResourceType$1(AuthHelper.scala:80) ~[kafka_2.13-4.0.0.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorizeByResourceType$1$adapted(AuthHelper.scala:79) ~[kafka_2.13-4.0.0.jar:?]
	at scala.Option.forall(Option.scala:420) ~[scala-library-2.13.15.jar:?]
	at kafka.server.AuthHelper.authorizeByResourceType(AuthHelper.scala:79) ~[kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaApis.handleInitProducerIdRequest(KafkaApis.scala:1535) [kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaApis.handle(KafkaApis.scala:185) [kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:158) [kafka_2.13-4.0.0.jar:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]

Then I added both to my config:

  authorization:
      type: custom
      delegateToKafkaAcls: true
      ......
   config:
      strimzi.authorization.delegate.to.kafka.acl: "true"
      ......

Then when I try I do not see any ACL related error. Please advise should we enable it or disable it?
I am aware that former is Kafka config while the latter is authorization config related for Keycloak.

Doubt-2:
When I try to follow the steps to confirm OAuth in action as mentioned here
I face errors.

My team-a.properties:

bootstrap.servers=my-cluster-kafka-bootstrap.kafka-operator.svc:9092

# Change this from SASL_PLAINTEXT to SASL_SSL if your listener name is 'tls'
# If you truly want no encryption, ensure the listener in Strimzi is type: plain
security.protocol=SASL_PLAINTEXT
sasl.mechanism=OAUTHBEARER

# Strimzi OAuth login callback handler
sasl.login.callback.handler.class=io.strimzi.kafka.oauth.client.JaasClientOauthLoginCallbackHandler

# JAAS configuration for obtaining access tokens
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required   oauth.token.endpoint.uri="http://sso.local.io:32080/realms/kafka-authz/protocol/openid-connect/token" \
  oauth.client.id="team-a-client" \
  oauth.client.secret="team-a-client-secret";

While producing a message on a_topic:

kafkaclientpod:/opt# $KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server my-cluster-kafka-bootstrap.kafka-operator.svc:9092   --topic a_topic   --producer.config team-a.properties
>hi
>[2025-12-29 18:47:02,523] ERROR [Producer clientId=console-producer] Aborting producer batches due to fatal error (org.apache.kafka.clients.producer.internals.Sender)
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
[2025-12-29 18:47:02,550] ERROR Error when sending message to topic a_topic with key: null, value: 2 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
^C[2025-12-29 18:47:06,886] ERROR [Producer clientId=console-producer] Error in kafka producer I/O thread while aborting transaction when during closing:  (org.apache.kafka.clients.producer.internals.Sender)
java.lang.IllegalStateException: Transactional method invoked on a non-transactional producer.
	at org.apache.kafka.clients.producer.internals.TransactionManager.ensureTransactional(TransactionManager.java:1091) ~[kafka-clients-4.0.0.jar:?]
	at org.apache.kafka.clients.producer.internals.TransactionManager.handleCachedTransactionRequestResult(TransactionManager.java:1208) ~[kafka-clients-4.0.0.jar:?]
	at org.apache.kafka.clients.producer.internals.TransactionManager.beginAbort(TransactionManager.java:326) ~[kafka-clients-4.0.0.jar:?]
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:277) [kafka-clients-4.0.0.jar:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]

No logs in Kafka broker pod logs. While keycloak pod has logs as:

2025-12-29 18:40:11,198 INFO  [org.keycloak.events] (executor-thread-6) type="PERMISSION_TOKEN", realmId="2325956d-c882-4755-9f38-d1c0f08a56e9", realmName="kafka-authz", clientId="team-a-client", userId="aa4c588a-bbc0-4f34-97e2-fe16f1e1f4e4", ipAddress="172.18.0.2", auth_method="oauth_credentials", audience="kafka", grant_type="urn:ietf:params:oauth:grant-type:uma-ticket", authSessionParentId="cb62bc64-b18d-4364-baac-d748a46f3909", authSessionTabId="JsDLIktC2yo"

2025-12-29 18:40:13,788 INFO  [org.keycloak.events] (executor-thread-6) type="CLIENT_LOGIN", realmId="2325956d-c882-4755-9f38-d1c0f08a56e9", realmName="kafka-authz", clientId="team-a-client", userId="0f1cc9c7-88f8-4d61-95fa-1ed67bb0cc83", ipAddress="10.244.2.1", token_id="trrtcc:7c81c135-b518-1b83-cd76-882017433f92", grant_type="client_credentials", scope="profile email", client_auth_method="client-secret", username="service-account-team-a-client", authSessionParentId="8046afb6-6bbd-43e4-8623-e71a9d7d9ac9", authSessionTabId="phnQDfn3bl8"

When I run this to get the JWT for team-a-client I see as what it is allowed to do:
cURL Request:

curl -s -X POST \
  -d "grant_type=urn:ietf:params:oauth:grant-type:uma-ticket&client_id=team-a-client&client_secret=team-a-client-secret&audience=kafka" \
  "http://sso.local.io:32080/realms/kafka-authz/protocol/openid-connect/token" | jq -R 'split(".") | .[1] | @base64d | fromjson' | jq .

cURL Response:

{
  "exp": 1767037317,
  "iat": 1767033717,
  "jti": "trrtna:06582e33-eb3f-9f4c-2bce-5a34eb11e53f",
  "iss": "http://sso.local.io:32080/realms/kafka-authz",
  "aud": "kafka",
  "sub": "0f1cc9c7-88f8-4d61-95fa-1ed67bb0cc83",
  "typ": "Bearer",
  "azp": "team-a-client",
  "acr": "1",
  "realm_access": {
    "roles": [
      "offline_access",
      "Dev Team A"
    ]
  },
  "resource_access": {
    "account": {
      "roles": [
        "manage-account",
        "manage-account-links",
        "view-profile"
      ]
    }
  },
  "authorization": {
    "permissions": [
      {
        "scopes": [
          "Write",
          "Alter",
          "Delete",
          "Describe",
          "Read",
          "Create",
          "DescribeConfigs",
          "AlterConfigs"
        ],
        "rsid": "4ab33901-0434-46b1-8017-d30aa02a4ae7",
        "rsname": "Topic:a_*"
      },
      {
        "scopes": [
          "Write",
          "Describe"
        ],
        "rsid": "fc245dde-92b5-4e4a-90b0-0c90715c31c3",
        "rsname": "Topic:x_*"
      },
      {
        "scopes": [
          "IdempotentWrite"
        ],
        "rsid": "274fa1fc-14b2-41d5-917b-945741706593",
        "rsname": "kafka-cluster:my-cluster,Cluster:*"
      },
      {
        "scopes": [
          "Describe",
          "Read"
        ],
        "rsid": "fc9214f5-507c-4417-a48b-3d9e11d80360",
        "rsname": "Group:a_*"
      }
    ]
  },
  "scope": "profile email",
  "email_verified": false,
  "preferred_username": "service-account-team-a-client"
}

If Dev Team A has full ownership for all topics starting with a_* then it should be able to produce messages on topic a_topic.

Btw, I have the topic:

Our AI friends have drawn this for me from kafka-authz.json:

Not sure what is missing in my config. Any hints regarding this would be very helpful and appreciated. Many thanks in advance!

[mstruk]

The following stacktrace:

2025-12-29 11:50:01 ERROR [data-plane-kafka-request-handler-5] KafkaApis:76 - [KafkaApi-0] Unexpected error handling request RequestHeader(apiKey=INIT_PRODUCER_ID, apiVersion=5, clientId=console-producer, correlationId=2, headerVersion=2) -- InitProducerIdRequestData(transactionalId=null, transactionTimeoutMs=2147483647, producerId=-1, producerEpoch=-1) with context RequestContext(header=RequestHeader(apiKey=INIT_PRODUCER_ID, apiVersion=5, clientId=console-producer, correlationId=2, headerVersion=2), connectionId='10.244.2.11:9092-127.0.0.6:34185-3-1', clientAddress=/127.0.0.6, principal=OAuthKafkaPrincipal(User:service-account-team-a-client, groups: null, session: 909467807, token: eyJh**BDtw), listenerName=ListenerName(PLAIN-9092), securityProtocol=SASL_PLAINTEXT, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=4.0.0), fromPrivilegedListener=false, principalSerde=Optional[io.strimzi.kafka.oauth.server.OAuthKafkaPrincipalBuilder@68849c7f])
java.lang.UnsupportedOperationException: Simple ACL delegation not enabled
	at io.strimzi.kafka.oauth.server.authorizer.KeycloakRBACAuthorizer.acls(KeycloakRBACAuthorizer.java:599) ~[kafka-oauth-keycloak-authorizer-0.17.0.jar:?]
	at org.apache.kafka.server.authorizer.Authorizer.authorizeByResourceType(Authorizer.java:217) ~[kafka-clients-4.0.0.jar:?]
	at io.strimzi.kafka.oauth.server.authorizer.KeycloakAuthorizer.authorizeByResourceType(KeycloakAuthorizer.java:218) ~[kafka-oauth-keycloak-authorizer-0.17.0.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorizeByResourceType$1(AuthHelper.scala:80) ~[kafka_2.13-4.0.0.jar:?]
	at kafka.server.AuthHelper.$anonfun$authorizeByResourceType$1$adapted(AuthHelper.scala:79) ~[kafka_2.13-4.0.0.jar:?]
	at scala.Option.forall(Option.scala:420) ~[scala-library-2.13.15.jar:?]
	at kafka.server.AuthHelper.authorizeByResourceType(AuthHelper.scala:79) ~[kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaApis.handleInitProducerIdRequest(KafkaApis.scala:1535) [kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaApis.handle(KafkaApis.scala:185) [kafka_2.13-4.0.0.jar:?]
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:158) [kafka_2.13-4.0.0.jar:?]
	at java.base/java.lang.Thread.run(Thread.java:840) [?:?]

Tells me that you are trying to list ACLs while using KeycloakAuthorizer with delegation disabled. That's in line with what you mention above how you configure the KeycloakAuthorizer.

Here's the code that throws the exception: https://github.com/strimzi/strimzi-kafka-oauth/blob/0.17.0/oauth-keycloak-authorizer/src/main/java/io/strimzi/kafka/oauth/server/authorizer/KeycloakRBACAuthorizer.java#L599

If you disable delegation then you can not list the standard ACLs as the listing command can not be delegated. It seems that the behaviour is correct. Usually when using KeycloakAuthorizer, you put all your users in Keycloak and you configure their permissions in Keycloak as described in the documentation (https://github.com/strimzi/strimzi-kafka-oauth/blob/0.17.1/examples/README-authz.md#using-keycloak-admin-console-to-configure-authorization).

Hopefully the configuration documentation is up-to-date and correct. If you find inconsistencies, please let us know.

[sahil-sharma]

Thank you @mstruk for your detailed response.
It seems to be okay now.
Though I have one small doubt before I mark this as an answer. I am asking here but if you think I should open a new discussion then would happily do the same.
When I test Kafka OAuth and try to produce a message to a_topic with team-a-client.properties with below command I get an error:

kafkaclientpod:/opt# $PRODUCER_SCRIPT --bootstrap-server $BOOTSTRAP_SERVER_ADDR --topic $A_TOPIC \
  --producer.config $TEAM_A_PROPERTIES_FILE
>hi
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
	at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:1113)
	at org.apache.kafka.clients.producer.internals.TransactionManager.maybeAddPartition(TransactionManager.java:402)
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:1018)
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:943)
	at org.apache.kafka.tools.ConsoleProducer.send(ConsoleProducer.java:112)
	at org.apache.kafka.tools.ConsoleProducer.loopReader(ConsoleProducer.java:101)
	at org.apache.kafka.tools.ConsoleProducer.start(ConsoleProducer.java:71)
	at org.apache.kafka.tools.ConsoleProducer.main(ConsoleProducer.java:60)
Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
[2026-01-08 18:42:01,306] ERROR [Producer clientId=console-producer] Aborting producer batches due to fatal error (org.apache.kafka.clients.producer.internals.Sender)
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
[2026-01-08 18:42:01,326] ERROR Error when sending message to topic a_topic with key: null, value: 2 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
kafkaclientpod:/opt# 

But when I do add this to producer command: --producer-property enable.idempotence=false it works fine.

kafkaclientpod:/opt# $PRODUCER_SCRIPT --bootstrap-server $BOOTSTRAP_SERVER_ADDR --topic $A_TOPIC \
  --producer.config $TEAM_A_PROPERTIES_FILE --producer-property enable.idempotence=false
>hi
>how 
>^Ckafkaclientpod:/opt# 

I do have this in my Keycloak permissions:

    {
      "name": "Cluster:*",
      "type": "Cluster",
      "ownerManagedAccess": false,
      "displayName": "",
      "attributes": {},
      "uris": [],
      "scopes": [
        {
          "name": "Describe"
        },
        {
          "name": "IdempotentWrite"
        },
        {
          "name": "DescribeConfigs"
        },
        {
          "name": "AlterConfigs"
        },
        {
          "name": "ClusterAction"
        }
      ],
      "icon_uri": ""
    },
--------------
    {
      "name": "Allow Teams to use Idempotent Producers",
      "description": "",
      "type": "resource",
      "logic": "POSITIVE",
      "decisionStrategy": "AFFIRMATIVE",
      "config": {
        "resources": "[\"Cluster:*\"]",
        "scopes": "[\"IdempotentWrite\"]",
        "applyPolicies": "[\"Dev Team B\",\"Dev Team A\"]"
      }
    }, 
    

To my understanding it should work without mentioning --producer-property enable.idempotence=false if above permissions are suffice.
I did try to temper with permissions like adding Describe, and DescribeConfigs after seeking help from our AI friends. But nothing worked.
Please advise if you spot an error or have an idea.

[sahil-sharma]

After a bit of debugging I found the root cause for this error and it was some permission that was overriding the legit ones.
Thank you for your time and effort to answer my doubts.

[mstruk]

When using custom authorization as you describe here:

 authorization:
      type: custom
      delegateToKafkaAcls: true
      ......
   config:
      strimzi.authorization.delegate.to.kafka.acl: "true"
      ......

The delegateToKafkaAcls has no effect as it is type: keycloak attribute. The other way to configure - under config - is the correct way.

[sahil-sharma]

Thank you for the info.

[mstruk]

No logs in Kafka broker pod logs.

You can enable Strimzi OAuth library logging for OAuth and Keycloak authorization by setting logging level for logger io.strimzi to DEBUG. How exactly depends on how you configured your logging overrides. If using log4j2.yaml you would add logger level configuration like this:

Configuration:
  ...
  Loggers:
    ...
    Logger:
      ...
      - name: io.strimzi
        level: DEBUG

[mstruk]

java.lang.IllegalStateException: Transactional method invoked on a non-transactional producer.

If you get this error while following the instructions to the letter, it might mean that the documentation is out of sync. I'll have to do a run through the instructions to see if I get the same error.

[sahil-sharma]

Luckily, I did not face this error.