Kafka mirror maker2 active-active replication issues .

[L040639]

We are running kafka active-passive in production, we are trying to work with active-active now , so below are my 2 mm2 configurations for 2 regions to replicate the topics .

I observed that , along with the mirror maker internal topics their replicated topics also created on both clusters side , i have added topics.exclude but still getting these duplications , is this the expected behaviour and the topic.exclude config is correct for the kafka active-active scenario because i want to avoid the internal topics to avoid the infinite loops .

**Example
mirrormaker2-cluster-configs-0
mirrormaker2-cluster-offsets-0
mirrormaker2-cluster-status-0

kafka-cluster.mirrormaker2-cluster-configs-0
kafka-cluster.mirrormaker2-cluster-offsets-0
kafka-cluster.mirrormaker2-cluster-status-0**

**Example
mirrormaker2-cluster-configs-0
mirrormaker2-cluster-offsets-0
mirrormaker2-cluster-status-0

kafka-cluster-dr.mirrormaker2-cluster-configs-0
kafka-cluster-dr.mirrormaker2-cluster-offsets-0
kafka-cluster-dr.mirrormaker2-cluster-status-0**

west to east mm2:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: mm2-west-east 
  namespace: kafka-streams-poc
spec:
  clusters:
    - alias: kafka-cluster-dr  ## source cluster
      authentication:
        clientId: <XXXX>
        clientSecret:
          key: secret
          secretName: entra-id-client
        scope: 'api://<XXX>/.default'
        tokenEndpointUri: 'https://login.microsoftonline.com/<XXXX>/oauth2/v2.0/token'
        type: oauth
      bootstrapServers: '<XXXX>' ## source cluster bootstrap address.
      tls:
        trustedCertificates:
          - certificate: ca.crt
            secretName: lk-ca-root-bundle
    - alias: kafka-cluster  ## target cluster
      authentication:
        clientId: <XXX>
        clientSecret:
          key: secret
          secretName: entra-id-client
        scope: 'api://<XXXX>.default'
        tokenEndpointUri: 'https://login.microsoftonline.com/<XXXX>/oauth2/v2.0/token'
        type: oauth
      bootstrapServers: '<XXXX>'   ## Target cluster bootstrap address.
      config:
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
      tls:
        trustedCertificates:
          - certificate: ca.crt
            secretName: lk-ca-root-bundle
  connectCluster: kafka-cluster
  mirrors:
    - checkpointConnector:
        config:
          checkpoints.topic.replication.factor: -1
          emit.checkpoints.interval.seconds: 60
          refresh.groups.interval.seconds: 600
          sync.group.offsets.enabled: 'true'
          sync.group.offsets.interval.seconds: 60
          topic.creation.default.partitions: 1
          topic.creation.default.replication.factor: 1
          replication.policy.separator: "."
          # replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
          replication.permitted.sources: "kafka-cluster-dr"  # Prevents infinite loop
        tasksMax: 1
      groupsPattern: .*
      sourceCluster: kafka-cluster-dr
      sourceConnector:
        config:
          offset-syncs.topic.replication.factor: -1
          offset-syncs.topic.location: target
          topic.creation.default.partitions: 1
          offset.lag.max: 100
          refresh.topics.interval.seconds: 600
          sync.topic.acls.enabled: 'false'
          topic.creation.default.replication.factor: 1
          replication.factor: -1
          topic.creation.enable: 'true'
          replication.policy.separator: "."
          # replication.policy.class: "io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy"
          replication.permitted.sources: "kafka-cluster-dr"  # Prevents infinite loop
          topics.exclude: "kafka-cluster.*|mirrormaker2-cluster-offsets-.*|mirrormaker2-cluster-status-.*|mirrormaker2-cluster-configs-.*|__.*|bootstrap.checkpoint|cleaner-offset-checkpoint|log-start-offset-checkpoint|meta.properties|recovery-point-offset-checkpoint|replication-offset-checkpoint"
        tasksMax: 1
      targetCluster: kafka-cluster
      topicsPattern: .*
  replicas: 1
  version: 3.9.0

east to west mm2

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: mm2-east-west
  namespace: kafka-streams-poc-mm
spec:
  clusters:
    - alias: kafka-cluster  ## source cluster
      authentication:
        clientId: <XXXX>
        clientSecret:
          key: secret
          secretName: entra-id-client
        scope: 'api://<XXXX>/.default'
        tokenEndpointUri: 'https://login.microsoftonline.com/<XXXX>/oauth2/v2.0/token'
        type: oauth
      bootstrapServers: '<XXXX>' ## source cluster bootstrap address.
      tls:
        trustedCertificates:
          - certificate: ca.crt
            secretName: lk-ca-root-bundle
    - alias: kafka-cluster-dr  ## target cluster
      authentication:
        clientId: <XXXX>
        clientSecret:
          key: secret
          secretName: entra-id-client
        scope: 'api://<XXXX>/.default'
        tokenEndpointUri: 'https://login.microsoftonline.com/<XXXX>/oauth2/v2.0/token'
        type: oauth
      bootstrapServers: '<XXXX>'  ## Target cluster bootstrap address.
      config:
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
      tls:
        trustedCertificates:
          - certificate: ca.crt
            secretName: lk-ca-root-bundle
  connectCluster: kafka-cluster-dr
  mirrors:
    - checkpointConnector:
        config:
          checkpoints.topic.replication.factor: -1
          emit.checkpoints.interval.seconds: 60
          refresh.groups.interval.seconds: 600
          sync.group.offsets.enabled: 'true'
          sync.group.offsets.interval.seconds: 60
          topic.creation.default.partitions: 1
          topic.creation.default.replication.factor: 1
          topic.creation.default.cleanup.policy: "compact"
          replication.policy.separator: "."
          # replication.policy.class: "io.strimzi.kafka.connect.mirror.DefaultReplicationPolicy"
          replication.permitted.sources: "kafka-cluster"  # Prevents infinite loop
        tasksMax: 1
      groupsPattern: .*
      sourceCluster: kafka-cluster
      sourceConnector:
        config:
          offset-syncs.topic.replication.factor: -1
          offset-syncs.topic.location: target
          topic.creation.default.partitions: 1
          offset.lag.max: 100
          refresh.topics.interval.seconds: 600
          sync.topic.acls.enabled: 'false'
          topic.creation.default.replication.factor: 1
          topic.creation.default.cleanup.policy: "compact"
          replication.factor: -1
          topic.creation.enable: 'true'
          replication.policy.separator: "."
          # replication.policy.prefix: "dr"
          # replication.policy.class: "io.strimzi.kafka.connect.mirror.DefaultReplicationPolicy"
          replication.permitted.sources: "kafka-cluster"  # Prevents infinite loop
          topics.exclude: "kafka-cluster-dr.*|mirrormaker2-cluster-offsets-.*|mirrormaker2-cluster-status-.*|mirrormaker2-cluster-configs-.*|__.*|bootstrap.checkpoint|cleaner-offset-checkpoint|log-start-offset-checkpoint|meta.properties|recovery-point-offset-checkpoint|replication-offset-checkpoint"
        tasksMax: 1
      targetCluster: kafka-cluster-dr
      topicsPattern: .*
  replicas: 1
  version: 3.9.0

[L040639]

@scholzj can someone clarify on this .

[tinaselenge]

@L040639 have you tried using topicsExcludePattern field of KafkaMirrorMaker2 CR?

https://strimzi.io/docs/operators/latest/configuring#type-KafkaMirrorMaker2MirrorSpec-reference

[tinaselenge]

The operator sets this config using topicsExcludePattern just like the topicsPattern you have used.

[tinaselenge]

but are you saying all the other patterns are successfully excluded except mirrormaker2-cluster-<> ones?

[L040639]

yes all the patterns excluded except mirror maker topics, that is the expected behaviour ? and i excluded some internal topics other than mirror maker topics , my below config is correct for active-active scenario ?

topics.exclude: "kafka-cluster.|mirrormaker2-cluster-offsets-.|mirrormaker2-cluster-status-.|mirrormaker2-cluster-configs-.|__.*|bootstrap.checkpoint|cleaner-offset-checkpoint|log-start-offset-checkpoint|meta.properties|recovery-point-offset-checkpoint|replication-offset-checkpoint

[tinaselenge]

I don't think that is expected. Very similar patterns seems to work fine for me. Are these other topics ones you created?
It might be useful look at the MM2 logs and check if this config took an effect definitely, for example:

	topic.filter.class = class org.apache.kafka.connect.mirror.DefaultTopicFilter
	topics = [.*]
	topics.exclude = [topic-example.*|topic-cde-.*|topic-abc-.*]

Side note, it is useful to name the internal topics prefixed with __ as these don't get replicated by the source connector by default (also included in your topic.exclude pattern). For example, in the connect config:

config.storage.topic: __mirrormaker2-cluster-configs
offset.storage.topic: __mirrormaker2-cluster-offsets
status.storage.topic: __mirrormaker2-cluster-status

Obviously not that useful in your case, since these topics are already created.

[L040639]

i am trying in my poc lab , i can revert my existing and try this and also do i need to mention the below internal topics aswell in the exclude config, otherwise there may be a chance of looping .

cleaner-offset-checkpoint|log-start-offset-checkpoint|meta.properties|recovery-point-offset-checkpoint|replication-offset-checkpoint