stream-client Cluster authorization failed

[veerendra2]

Hello, One of our developer facing issue getting Cluster authorization failed while using kafka streams java SDK.

Kafka Version: 2.6.0
Openshift: 3.11
Strimzi: v0.20.1

Below is the error from the java code

...
2022-05-09 09:56:44.029 ERROR 1 --- [read-1-producer] o.a.k.clients.producer.internals.Sender  : [Producer clientId=trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af-StreamThread-1-producer] Aborting producer batches due to fatal error

org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.

2022-05-09 09:56:44.031 ERROR 1 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af] Encountered the following exception during processing and the registered exception handler opted to SHUTDOWN_CLIENT. The streams client is going to shut down now. 

org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic trade-bd-writer-qas-dummys-qas-STATE-STORE-0000000000-changelog for task 0_0 due to:
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
Written offsets would not be recorded and no more records would be sent since this is a fatal error.
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:209) ~[kafka-streams-3.0.1.jar!/:na]
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:196) ~[kafka-streams-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1350) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:273) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:161) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:773) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.maybeAbortBatches(Sender.java:498) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:307) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:243) ~[kafka-clients-3.0.1.jar!/:na]
	at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]
Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.

2022-05-09 09:56:44.031  INFO 1 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af] State transition from RUNNING to PENDING_ERROR
2022-05-09 09:56:44.031 DEBUG 1 --- [-StreamThread-1] c.s.b.c.a.k.KafkaStreamStateListener     : Stream State transition from RUNNING to PENDING_ERROR
2022-05-09 09:56:44.032  INFO 1 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN
2022-05-09 09:56:44.032  INFO 1 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af-StreamThread-1] Shutting down
2022-05-09 09:56:44.032  INFO 1 --- [ms-close-thread] o.a.k.s.p.internals.StreamThread         : stream-thread [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af-StreamThread-1] Informed to shut down
2022-05-09 09:56:44.034 ERROR 1 --- [-StreamThread-1] o.a.k.s.processor.internals.TaskManager  : stream-thread [trade-bd-writer-qas-a44b6cb3-e4c6-4101-bf70-4e35f46ee7af-StreamThread-1] Error flushing caches of dirty task 0_0 

org.apache.kafka.streams.errors.StreamsException: Error encountered sending record to topic trade-bd-writer-qas-dummys-qas-STATE-STORE-0000000000-changelog for task 0_0 due to:
org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.
Written offsets would not be recorded and no more records would be sent since this is a fatal error.
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:209) ~[kafka-streams-3.0.1.jar!/:na]
	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:196) ~[kafka-streams-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1350) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:273) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:161) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:773) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.maybeAbortBatches(Sender.java:498) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:307) ~[kafka-clients-3.0.1.jar!/:na]
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:243) ~[kafka-clients-3.0.1.jar!/:na]
	at java.base/java.lang.Thread.run(Unknown Source) ~[na:na]
Caused by: org.apache.kafka.common.errors.ClusterAuthorizationException: Cluster authorization failed.

At kafka broker

...
{"@timestamp":"2022-05-02T08:33:12.403Z","source_host":"test-dev-kafka-2","file":"AclAuthorizer.scala","method":"logAuditMessage","level":"INFO","line_number":"477","thread_name":"data-plane-kafka-request-handler-3","@version":1,"logger_name":"kafka.authorizer.logger","message":"Principal = User:CN=trade-bd-writer-qas is Denied Operation = IdempotentWrite from host = 10.128.6.1 on resource = Cluster:LITERAL:kafka-cluster for request = InitProducerId with resourceRefCount = 1","class":"kafka.security.authorizer.AclAuthorizer","mdc":{}}
...

KafkaUser definition

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  name: trade-bd-writer-qas
  labels:
    strimzi.io/cluster: test-dev
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Consumer Acls for topic dummy_assignments-qas
      - resource:
          type: topic
          name: dummy_assignments-qas
          patternType: literal
        operation: Read
        host: "*"
      - resource:
          type: topic
          name: dummy_assignments-qas
          patternType: literal
        operation: Describe
        host: "*"
      # Consumer Acls for topic dummys-qas
      - resource:
          type: topic
          name: dummys-qas
          patternType: literal
        operation: Read
        host: "*"
      - resource:
          type: topic
          name: dummys-qas
          patternType: literal
        operation: Describe
        host: "*"
      # Producer Acls for topic dummy_assignments-qas.DLT
      - resource:
          type: topic
          name: dummy_assignments-qas.DLT
          patternType: literal
        operation: Write
        host: "*"
      - resource:
          type: topic
          name: dummy_assignments-qas.DLT
          patternType: literal
        operation: Read
        host: "*"
      - resource:
          type: topic
          name: dummy_assignments-qas.DLT
          patternType: literal
        operation: Create
        host: "*"
      - resource:
          type: topic
          name: dummy_assignments-qas.DLT
          patternType: literal
        operation: Describe
        host: "*"
      # Producer Acls for STATE_STORE topics used by kafka streams
      - resource:
          type: topic
          name: trade-bd-writer-qas-dummys-qas-STATE-STORE
          patternType: prefix
        operation: All
        host: "*"
      - resource:
          type: topic
          name: trade-bd-writer-qas-unprocessed_dummy_assignments-qas
          patternType: prefix
        operation: All
        host: "*"
      # Consumer group trade-bd-writer
      - resource:
          type: group
          name: trade-bd-writer-qas
          patternType: literal
        operation: Read
        host: "*"

We are not quite sure what was issue at ACL
Anyone has similar issue? or any input would be helpful

Thanks

[scholzj]

Well, as described in the error message - you seem to be missing the ACL right for IdempotentWrite on the Cluster resource. At least I don't see it in your ACLs anywhere. Or did I missed it somewhere?

[veerendra2]

But I have added rule

...
      - resource:
          type: topic
          name: trade-bd-writer-qas-dummy-qas-STATE-STORE
          patternType: prefix
        operation: All
        host: "*"
...

With operations :All. I thought it should work

[veerendra2]

Thanks, that worked.

...
      - resource:
          type: cluster
          name: "*"
          name: kafka-cluster
          patternType: literal
        operation: All
        operation: IdempotentWrite
        host: "*"
      - resource:
          type: cluster
          name: kafka-cluster
          patternType: literal
        operation: Create
        host: "*"
      - resource:
          type: cluster
          name: kafka-cluster
          patternType: literal
        operation: Describe
        host: "*"
...

[scholzj]

Glad it works now.

With operations :All. I thought it should work

It is operation: All, but the resource type is topic. So it will do all operations on topics, but not on cluster.