Bidirectional data replication not happening in strimzi mirrormaker2

[Raju-kf]

Hello, I am using Strimzi MirrorMaker2 for replicating data between two kafka clusters. Bidirectional data replication is not happening in my case. I'm using strimzi operator 0.26 and kafka 2.8.0. Please let me know any special configuration required to enable bidirectional data replication in following mirrormaker2 configurations. Please suggest any changes and fixes. Thanks.

->From source to target cluster able to replicated topics and data but from target to source data replication is not happening.
->Topics are getting duplicated in both kafka clusters.
-> How can we configure data flow policies from A -> B and B->A clusters, any example configuration file is there.

Providing my mirrormaker.yaml file:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
 name: my-mirror-maker-2
spec:
 version: 2.8.0
 replicas: 1
 connectCluster: "my-target-cluster"
 clusters:
 - alias: "my-source-cluster"
   bootstrapServers: xxxx:30056
   tls:
     trustedCertificates:
       - secretName: my-cluster-cluster-ca-cert
         certificate: ca.crt
 - alias: "my-target-cluster"
   bootstrapServers: xxxx:30054
   tls:
     trustedCertificates:
       - secretName: my-target-cluster-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-source-cluster"
   targetCluster: "my-target-cluster"
   sourceConnector:
     config:
       replication.factor: 1
       offset-syncs.topic.replication.factor: 1
       sync.topic.acls.enabled: "false"
       refresh.topics.interval.seconds: 5
   heartbeatConnector:
     config:
       heartbeats.topic.replication.factor: 1
   checkpointConnector:
     config:
       checkpoints.topic.replication.factor: 1
   topicsPattern: ".*"
   groupsPattern: ".*"
 resources:
   requests:
     cpu: 200m
     memory: 2Gi
   limits:
     cpu: 250m
     memory: 2Gi

Topics list from source kafka c;usters:

[kafka@my-cluster-kafka-0 kafka]$ ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
__consumer_offsets
__strimzi-topic-operator-kstreams-topic-store-changelog
__strimzi_store_topic
cluster-c1
cluster-c1-1
heartbeats
mirrormaker2-cluster-configs
mirrormaker2-cluster-offsets
mirrormaker2-cluster-status
mm2-offset-syncs.my-target-cluster.internal
my-cluster-connect-cluster-configs
my-cluster-connect-cluster-offsets
my-cluster-connect-cluster-status
my-source-cluster.checkpoints.internal
my-source-cluster.cluster-c2
my-source-cluster.heartbeats
my-source-cluster.mirrormaker2-cluster-configs
my-source-cluster.mirrormaker2-cluster-offsets
my-source-cluster.mirrormaker2-cluster-status
my-source-cluster.my-source-cluster.cluster-c1
my-source-cluster.my-source-cluster.cluster-c1-1
my-source-cluster.my-source-cluster.heartbeats
my-source-cluster.my-source-cluster.mirrormaker2-cluster-configs
my-source-cluster.my-source-cluster.mirrormaker2-cluster-offsets
my-source-cluster.my-source-cluster.mirrormaker2-cluster-status
my-source-cluster.my-source-cluster.my-cluster-connect-cluster-configs
my-source-cluster.my-source-cluster.my-cluster-connect-cluster-offsets
my-source-cluster.my-source-cluster.my-cluster-connect-cluster-status
my-source-cluster.my-source-cluster.my-source-cluster.cluster-c2
my-source-cluster.my-source-cluster.my-source-cluster.heartbeats
my-source-cluster.my-source-cluster.my-source-cluster.mirrormaker2-cluster-configs
my-source-cluster.my-source-cluster.my-source-cluster.mirrormaker2-cluster-offsets
my-source-cluster.my-source-cluster.my-source-cluster.mirrormaker2-cluster-status
my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.cluster-c1
my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.cluster-c1-1

Mirrormaker pod logs:

2021-11-08 15:39:24,512 INFO Found 0 new topic-partitions on my-source-cluster. Found 0 deleted topic-partitions on my-source-cluster. Found 41 topic-partitions missing on my-target-cluster. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [Scheduler for MirrorSourceConnector-refreshing topics]
2021-11-08 15:39:24,517 INFO refreshing topics took 21 ms (org.apache.kafka.connect.mirror.Scheduler) [Scheduler for MirrorSourceConnector-refreshing topics]
2021-11-08 15:39:24,518 INFO SourceConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = my-source-cluster->my-target-cluster.MirrorSourceConnector
predicates = []
tasks.max = 1
topic.creation.groups = []
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.SourceConnectorConfig) [DistributedHerder-connect-1-1]
2021-11-08 15:39:24,518 INFO EnrichedConnectorConfig values:
config.action.reload = restart
connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
errors.log.enable = false
errors.log.include.messages = false
errors.retry.delay.max.ms = 60000
errors.retry.timeout = 0
errors.tolerance = none
header.converter = null
key.converter = null
name = my-source-cluster->my-target-cluster.MirrorSourceConnector
predicates = []
tasks.max = 1
topic.creation.groups = []
transforms = []
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig$EnrichedConnectorConfig) [DistributedHerder-connect-1-1]
2021-11-08 15:39:24,519 WARN Could not create topic my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.target-topic. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-13]
org.apache.kafka.common.errors.InvalidTopicException: Topic name is illegal, it can't be longer than 249 characters, topic name: my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.target-topic
2021-11-08 15:39:24,519 WARN Could not create topic my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.mirrormaker2-cluster-offsets. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-13]
org.apache.kafka.common.errors.InvalidTopicException: Topic name is illegal, it can't be longer than 249 characters, topic name: my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.mirrormaker2-cluster-offsets
2021-11-08 15:39:24,519 WARN Could not create topic my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.source-topic-nov-1. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-13]
org.apache.kafka.common.errors.InvalidTopicException: Topic name is illegal, it can't be longer than 249 characters, topic name: my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.source-topic-nov-1
2021-11-08 15:39:24,519 WARN Could not create topic my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.heartbeats. (org.apache.kafka.connect.mirror.MirrorSourceConnector) [kafka-admin-client-thread | adminclient-13]
org.apache.kafka.common.errors.InvalidTopicException: Topic name is illegal, it can't be longer than 249 characters, topic name: my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.my-source-cluster.heartbeats

[scholzj]

You should normally use 2 Mirror Makers => one for direction A->B and one for direction B->A.

From your list of topics, it almost looks like it is cycling. I wonder if you have 2 Mirror Makers using cluster aliases my-source-cluster and my-target-cluster aliases. But maybe you do not use the same alias for the same cluster in both instances? For bi-directional mirroring, it is better to use aliases such as cluster-A and cluster-B and make sure they always point to the right cluster. That should help you to avoid the cycling and should properly mirror the topics with their own prefixes.

[Raju-kf]

Thanks for the reply. I tried and deployed one mirrormaker for each cluster with following configurations. Now I could see no topic is being created in target cluster when I create topic from source cluster. Please suggest anything wrong in my deployment files.

Source-mm2.yaml :

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 2.8.0
  replicas: 1
  connectCluster: "mm-cluster-c1"
  clusters:
  - alias: "cluster-A"
    bootstrapServers: 164.88.15.152:30056
    tls:
      trustedCertificates:
        - secretName: my-cluster-cluster-ca-cert
          certificate: ca.crt
  - alias: "cluster-B"
    bootstrapServers: 164.88.15.152:30054
    tls:
      trustedCertificates:
        - secretName: my-target-cluster-cluster-ca-cert
          certificate: ca.crt
    config:
      config.storage.replication.factor: -1
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "cluster-A"
    targetCluster: "cluster-B"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 5
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: ".*"
    groupsPattern: ".*"
  resources:
    requests:
      cpu: 200m
      memory: 2Gi
    limits:
      cpu: 250m
      memory: 2Gi

Target-mm2.yaml:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 2.8.0
  replicas: 1
  connectCluster: "mm-cluster-c2"
  clusters:
  - alias: "cluster-A"
    bootstrapServers: 164.88.15.152:30056
    tls:
      trustedCertificates:
        - secretName: my-cluster-cluster-ca-cert
          certificate: ca.crt
  - alias: "cluster-B"
    bootstrapServers: 164.88.15.152:30054
    tls:
      trustedCertificates:
        - secretName: my-target-cluster-cluster-ca-cert
          certificate: ca.crt
    config:
      config.storage.replication.factor: -1
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "cluster-A"
    targetCluster: "cluster-B"
    sourceConnector:
      config:
        replication.factor: 1
        offset-syncs.topic.replication.factor: 1
        sync.topic.acls.enabled: "false"
        refresh.topics.interval.seconds: 5
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 1
    checkpointConnector:
      config:
        checkpoints.topic.replication.factor: 1
    topicsPattern: ".*"
    groupsPattern: ".*"
  resources:
    requests:
      cpu: 200m
      memory: 2Gi
    limits:
      cpu: 250m
      memory: 2Gi

[scholzj]

I see two issues with these files ...

• The connectCluster needs to be always the target cluster. So in one case cluster-A and in one case cluster-B
• The source and target cluster are in both cases the same. You need to have them the other way around in each YAML => once A->B and once B->A