Mirror Maker 2 error : "ERROR Scheduler for MirrorSourceConnector caught exception in scheduled task: syncing topic configs"

[fsouza-bux]

Hi,

We have been using kafka mirror maker 2 with the following configuration:

- TARGET: strimzi 0.21.1 -> kafka 2.7.0
- SOURCE: kafka 2.6.0 (remote to where we are running mirror)

On the TARGET side we are running strimzi with an ADMIN user, but on the SOURCE side it's a user with READ / DESCRIBE, and sometimes WRITE to topics, but not to all of them.

In this case we are testing a single topic, with the user being used capable of READ/WRITE/DESCRIBE the topic.

I keep anyway seeing the following error on mirror running:

2021-03-08 08:23:44,095 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:397)
	at org.apache.kafka.connect.mirror.MirrorSourceConnector.syncTopicConfigs(MirrorSourceConnector.java:298)
	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.

I did a couple of tests regarding the config of the topic being mirrored:

• Didn't create the topic on TARGET ahead, but the config of the TOPIC was not mirrored, as I could see later in the CRD pushed by the topic-operator, so config in the CRD was empty.. PS: sync.topics.configs.enabled was not set at this moment.
• Creating the topic in advance, still throws the error message in the log.
• I added sync.topics.configs.enabled as false, to try and prevent those messages to show up when creating the topic on our side but, it's still showing up.
• When starting the mirror, on TARGET we can see all the topics for config, offset and so on being created without any errors.

Mirror maker seems to still be working though, as we got some messages, very few, and the following message as INFO can be seen in the logs:

2021-03-08 08:23:44,114 INFO Found 0 new topic-partitions on <SOURCE>. Found 0 deleted topic-partitions on <SOURCE>. Found 1 topic-partitions missing on <TARGET>. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [Scheduler for MirrorSourceConnector-refreshing topics]

On the SOURCE side we have create the topic mm2-offset-syncs.<TARGET-ALIAS>.internal manually, as the user only has READ/WRITE/DESCRIBE.

Here's the mirror spec:

apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaMirrorMaker2
metadata:
  name: kafka-mirror
spec:
  version: 2.7.0
  replicas: 1
  connectCluster: "<TARGET>"
  clusters:
    - alias: "<SOURCE>"
      authentication:
        type: plain
        username: <USER>
        passwordSecret:
          password: password
          secretName: <USER>
      bootstrapServers: <SOURCE_BOOTSTRAP_SERVERS>
      config:
        ssl.endpoint.identification.algorithm: ""
    - alias: "<TARGET>"
      authentication:
        passwordSecret:
          password: password
          secretName: <ADMIN-USER>
        type: scram-sha-512
        username: <ADMIN-USER>
      bootstrapServers: <TARGET_BOOTSTRAP_SERVERS>
  mirrors:
    - sourceCluster: "<SOURCE>"
      targetCluster: "<TARGET>"
      sourceConnector:
        config:
          replication.policy.class: io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy
          replication.policy.separator: ""
          sync.topic.acls.enabled: "false"
          sync.topics.configs.enabled: "false"
      # heartbeatConnector:
      #   config:
      #     heartbeats.topic.replication.factor: 1
      # checkpointConnector:
      #   config:
      #     checkpoints.topic.replication.factor: 1
      #     replication.policy.class: io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy
      #     replication.policy.separator: ""
      topicsPattern: "TOPIC_EXAMPLE"
      groupsPattern: ".*"
  resources:
    limits:
      memory: 2.5Gi
    requests:
      memory: 2Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: "INFO"
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    -Xms: 1g
    -Xmx: 2g
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        key: metrics-config.yml
        name: mirror-maker-2-metrics

So a couple of questions then:

• Is the different versions in kafka causing the issue and not bringing the config from source?
• Is any configuration maybe missing? What we want to achieve is an ACTIVE->PASSIVE configuration for mirroring from 2.6.0 kafka to our new strimzi environment.
• Is the message because of some missing permissions on SOURCE, or even on TARGET? I could not really figure out or find if this message is really related to one side or the other.

Thanks,

Fernando

[fsouza-bux]

Hi all, for the ACL error above we got to fix it with giving the user DescribeConfig ACL on the SOURCE cluster in the target topic, so the access to get the config is solved. The user itself has only Read, Describe and DescribeConfig (It has Write as well but i don't think this matter for this exercise).
Though we see the following message on our logs as INFO:

2021-03-09 09:02:20,813 INFO Found 0 new topic-partitions on <SOURCE>. Found 0 deleted topic-partitions on <SOURCE>. Found 1 topic-partitions missing on <TARGET>. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [Scheduler for MirrorSourceConnector-refreshing topics]

Not sure though what this INFO message means, or if we really need to worry about it, but the part saying Found 1 topic-partitions missing on worries me a little bit.

Does anyone has a take on what it is? @ajborley, sorry to tag you here, but I saw you replying some questions regarding mm2 here in strimzi. Would you know the reason of the message saying that there's a partition missing on target?