Getting topic authorization error with MM2 : org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.

[Gulnaz2021]

Hi,

I've deployed 3 clusters with authorization type simple, tls enabled with Strimzi operator version20, kafka application version 2.6.0 and trying the below scenario:

1. Cluster1(topic1)-producing and consuming messages, MM2+Cluster2(mirrored-topic1)-consuming messages
2. Cluster2(mirrored-topic1)-producing and consuming messages, MM2+ Cluster3(mirrored-mirrored-topic1)-consuming messages,
   So, in the 2) approach I've observed topic authorization errors in the MM2 logs specific to the topics, though all the produced messages were getting consumed in the mirrored topic,
   Error:
   2021-03-23 13:52:24,935 WARN Could not create topic-partitions for my-topic2. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
   org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.
   2021-03-23 13:52:24,935 WARN Could not create topic-partitions for mirrormaker2-cluster-status. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
   org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.
   2021-03-23 13:52:24,935 WARN Could not create topic-partitions for heartbeats. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
   org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.
   2021-03-23 13:52:24,935 WARN Could not create topic-partitions for strimzi.cruisecontrol.metrics. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
   org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.
   2021-03-23 13:52:24,935 WARN Could not create topic-partitions for mirrormaker2-cluster-offsets. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
   org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.

And for 2)approach, when deployed the cluster with superuser, again observed below error in MM2 logs, though all the messages were getting consumed from the mirrored topic.

2021-03-24 12:29:07,767 ERROR Scheduler for MirrorSourceConnector caught exception in scheduled task: syncing topic configs (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-syncing topic configs]
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Topic authorization failed.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
at org.apache.kafka.connect.mirror.MirrorSourceConnector.describeTopicConfigs(MirrorSourceConnector.java:375)
at org.apache.kafka.connect.mirror.MirrorSourceConnector.syncTopicConfigs(MirrorSourceConnector.java:276)
at org.apache.kafka.connect.mirror.Scheduler.run(Scheduler.java:93)
at org.apache.kafka.connect.mirror.Scheduler.executeThread(Scheduler.java:112)
at org.apache.kafka.connect.mirror.Scheduler.lambda$scheduleRepeating$0(Scheduler.java:50)
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.common.errors.TopicAuthorizationException: Topic authorization failed.

Code added for the superuser in the cluster:
authorization:
type: simple
superUsers:
- CN=(source cluster user)
- CN=(target cluster user)
- CN=root
So, is there any specific configuration, I need to update. Can I please get help in this regard.

[scholzj]

Can you share the full YAMLs of the Kafka clusters and the MirrorMakers? Why do you have (source cluster user) and (target cluster user) in the same list? Shouldn't they be each on a different cluster? One on target and one on source?

[Gulnaz2021]

Can you share the full YAMLs of the Kafka clusters and the MirrorMakers?

For Kafka cluster below yaml file has been used, with difference in the metadata name, and resources, and when used for the target cluster- added the superuser section.

  apiVersion: kafka.strimzi.io/v1beta1
  kind: Kafka
  metadata:
    name: my-cluster
  spec:
    kafka:
      version: 2.6.0
      replicas: 3
      resources:
        requests:
          memory: 4Gi
          cpu: "4"
        limits:
          memory: 4Gi
          cpu: "4"
      listeners:
        - name: plain
          port: 9092
          type: internal
          tls: false
        - name: tls
          port: 9093
          type: internal
          tls: true
          authentication:
            type: tls
        - name: external
          port: 9094
          type: loadbalancer
          tls: true
          authentication:
            type: tls
      authorization:
        type: simple
      template:
        externalBootstrapService:
          metadata:
            annotations:
              cloud.google.com/load-balancer-type: "Internal"
        perPodService:
          metadata:
            annotations:
              cloud.google.com/load-balancer-type: "Internal"
      config:
        offsets.topic.replication.factor: 3
        transaction.state.log.replication.factor: 3
        transaction.state.log.min.isr: 2
        log.message.format.version: "2.6"
      storage:
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
      metrics: {}
      jvmOptions:
        -Xms: 2048m
        -Xmx: 2048m
        gcLoggingEnabled: true
    zookeeper:
      replicas: 3
      storage:
        type: persistent-claim
        size: 50Gi
        deleteClaim: false
      resources:
        requests:
          memory: 4Gi
          cpu: "4"
        limits:
          memory: 4Gi
          cpu: "4"
      metrics: {}
    entityOperator:
      topicOperator: {}
      userOperator: {}
    kafkaExporter:
      topicRegex: ".*"
      groupRegex: ".*"
    cruiseControl: {}

MM2 yaml file:
  apiVersion: kafka.strimzi.io/v1alpha1
  kind: KafkaMirrorMaker2
  metadata:
    name: mirrormaker1
  spec:
    version: 2.6.0
    replicas: 1
    connectCluster: "my-cluster-target"
    clusters:
    - alias: "my-cluster-src"
      authentication:
        certificateAndKey:
          certificate: user.crt
          key: user.key
          secretName: source-user
        type: tls
      bootstrapServers: my-cluster-src-kafka-bootstrap.<ns>.svc:9093
      tls:
        trustedCertificates:
          - secretName: my-cluster-src-cluster-ca-cert
            certificate: ca.crt
    - alias: "my-cluster-target"
      authentication:
        certificateAndKey:
          certificate: user.crt
          key: user.key
          secretName: target-user
        type: tls
      bootstrapServers: my-cluster-target-kafka-bootstrap.<ns>.svc:9093
      tls:
        trustedCertificates:
          - secretName: my-cluster-target-cluster-ca-cert
            certificate: ca.crt
      config:
        config.storage.replication.factor: 1
        offset.storage.replication.factor: 1
        status.storage.replication.factor: 1
    mirrors:
    - sourceCluster: "my-cluster-src"
      targetCluster: "my-cluster-target"
      sourceConnector:
        config:
          replication.factor: 1
          offset-syncs.topic.replication.factor: 3
          sync.topic.acls.enabled: "false"
          replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
      heartbeatConnector:
        config:
          heartbeats.topic.replication.factor: 1
      checkpointConnector:
        config:
          checkpoints.topic.replication.factor: 1
      topicsPattern: ".*"
      groupsPattern: ".*"

Why do you have (source cluster user) and (target cluster user) in the same list? Shouldn't they be each on a different cluster? One on target and one on source?

Yes they are on different clusters only, but as we've been getting topic authorization error, so tried by adding the above mentioned superuser section in the target cluster. Let me know, if the mentioned configuration is wrong.

[scholzj]

Well, the YAML does not contain the super users. So it is not clear if you got the usernames right etc. Also, the ACL error might not be just from the target cluster, the ACLs need to be set right on the source cluster as well.

[Gulnaz2021]

There are 2 ways, I've tried to test when got the ACL error-
First, the above shared yaml file for kafka cluster has been used for all the 3 clusters and got the topic authorization errors, such as :
2021-03-23 13:52:24,935 WARN Could not create topic-partitions for my-topic2. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed.
2021-03-23 13:52:24,935 WARN Could not create topic-partitions for mirrormaker2-cluster-status. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-14]
org.apache.kafka.common.errors.TopicAuthorizationException: The topic authorization is failed...…..

Second, used the above shared yaml file for kafka cluster:
for scenario 1)Cluster1(topic1)-producing and consuming messages, MM2+Cluster2(mirrored-topic1)-consuming messages,
and the below yaml file was used for the Cluster3, in scenario 2) Cluster2(mirrored-topic1)-producing and consuming messages, MM2+ Cluster3(mirrored-mirrored-topic1)-consuming messages

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
name: my-cluster
spec:
kafka:
version: 2.6.0
replicas: 3
resources:
requests:
memory: 4Gi
cpu: "4"
limits:
memory: 4Gi
cpu: "4"
listeners:
- name: plain
port: 9092
type: internal
tls: false
- name: tls
port: 9093
type: internal
tls: true
authentication:
type: tls
- name: external
port: 9094
type: loadbalancer
tls: true
authentication:
type: tls
authorization:
type: simple
superUsers:
- CN=(source cluster user)
- CN=(target cluster user)
- CN=root
template:
externalBootstrapService:
metadata:
annotations:
cloud.google.com/load-balancer-type: "Internal"
perPodService:
metadata:
annotations:
cloud.google.com/load-balancer-type: "Internal"
config:
offsets.topic.replication.factor: 3
transaction.state.log.replication.factor: 3
transaction.state.log.min.isr: 2
log.message.format.version: "2.6"
storage:
type: persistent-claim
size: 100Gi
deleteClaim: false
metrics: {}
jvmOptions:
-Xms: 2048m
-Xmx: 2048m
gcLoggingEnabled: true
zookeeper:
replicas: 3
storage:
type: persistent-claim
size: 50Gi
deleteClaim: false
resources:
requests:
memory: 4Gi
cpu: "4"
limits:
memory: 4Gi
cpu: "4"
metrics: {}
entityOperator:
topicOperator: {}
userOperator: {}
kafkaExporter:
topicRegex: "."
groupRegex: "."
cruiseControl: {}

And got the error as :
2021-03-24 12:29:07,767 ERROR Scheduler for MirrorSourceConnector caught exception in scheduled task: syncing topic configs (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-syncing topic configs]
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicAuthorizationException: Topic authorization failed.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
........................

[scholzj]

Sorry, I'm afraid your scenario is too complicated for me. From the Mirror Maker errors, it is not completely clear where the error is.

The ACL errors are normally tracked on the brokers which much more descriptive message. So it might be best to track these errors in the broker logs where they should clearly log also the user which had the ACL error. So that should make it easier to compare with the superUser list.