Messages replicating constantly

[AEncinas19]

Hello,
I'm using an active/active replication with MirrorMaker. In order to do this, I've created two MakerMirror2, one that replicates from my-cluster to my-cluster 2 and another for the other direction.

When I write a message from one Kafka pod in a cluster, it replicates into the other cluster. However I don't know why the messages are continuosly generating copied data.

Here are my MirrorMaker yamls:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2
spec:
  version: 3.3.1
  replicas: 1
  connectCluster: "my-cluster2"
  clusters:
  - alias: "my-cluster"
    bootstrapServers: my-cluster-kafka-bootstrap:9092
  - alias: "my-cluster2"
    bootstrapServers: my-cluster2-kafka-bootstrap:9092
    config:
      # -1 means it will use the default replication factor configured in the broker
      # Number of replicas configuration for each message at target cluster.
      config.storage.replication.factor: 1
      # Number of replicas configuration for each topic at target cluster.
      offset.storage.replication.factor: -1
      status.storage.replication.factor: -1
  mirrors:
  - sourceCluster: "my-cluster"
    targetCluster: "my-cluster2"
    sourceConnector:
      config:
        # Replication factor for new topics (default 2)
        replication.factor: 3
        # Replication factor for the internal offset-syncs topic (default 3). The offset syncs topic maps the source and target offsets for replicated topic partitions.
        offset-syncs.topic.replication.factor: 3
        sync.topic.acls.enabled: "false"
        # Frecuency of topic refresh (default 600)
        refresh.topics.interval.seconds: 60
        replication.policy.separator: ""
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
      # Replication factor for the internal checkpoints topic (default 3). The checkpoint topic maps the last commited offset in the source and target cluster for replicated topic partitions in each consumer group.
        checkpoints.topic.replication.factor: 3
        replication.policy.separator: ""
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    topicsPattern: "flows"
    groupsPattern: ".*"

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker-2-active
spec:
  version: 3.3.1
  replicas: 1
  connectCluster: "my-cluster"
  clusters:
  - alias: "my-cluster2"
    bootstrapServers: my-cluster2-kafka-bootstrap:9092
  - alias: "my-cluster"
    bootstrapServers: my-cluster-kafka-bootstrap:9092
    config:
      # -1 means it will use the default replication factor configured in the broker
      config.storage.replication.factor: 1
      offset.storage.replication.factor: 1
      status.storage.replication.factor: 1
  mirrors:
  - sourceCluster: "my-cluster2"
    targetCluster: "my-cluster"
    sourceConnector:
      config:
        # Replication factor for new topics (default 2)
        replication.factor: 3
        # Replication factor for the internal offset-syncs topic (default 3). The offset syncs topic maps the source and target offsets for replicated topic partitions.
        offset-syncs.topic.replication.factor: 3
        sync.topic.acls.enabled: "false"
        # Frecuency of topic refresh (default 600)
        refresh.topics.interval.seconds: 60
        replication.policy.separator: ""
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    heartbeatConnector:
      config:
        heartbeats.topic.replication.factor: 3
    checkpointConnector:
      config:
      # Replication factor for the internal checkpoints topic (default 3). The checkpoint topic maps the last commited offset in the source and target cluster for replicated topic partitions in each consumer group.
        checkpoints.topic.replication.factor: 3
        replication.policy.separator: ""
        replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
    topicsPattern: "flows"
    groupsPattern: ".*"

The output of the consumer is the following:

[scholzj]

This is because you use the IdentityReplicationPolicy. You cannot do it when you do active-active mirroring. There is no real active-active mirroring in Mirror Maker ... you normally just rename the topics, but an individual topic is always mirrored only in one direction. When you try to do it with the IdentityReplicationPolicy, then this is what happens. The IdentityReplicationPolicy is normally designed for active-passive mirroring where the passive node basically works as a kind of backup and not an active site where messages are produced.

[scholzj]

PS: Keep in mind that even without the messages being copied again and again, you lose a lot of functionality. The message order will be messed up, which also makes it really hard to move clients between the clusters, etc. So I think the usefulness would be very limited.

[AEncinas19]

Yes, I've read the documentation and I thought that you can use the same topics in different clusters in order to be able to aggregate data from differents sources geographically separated and scale it.

[scholzj]

Again, that depends - aggregate data from differents sources geographically separated - would often mean pulling from some regional clusters to a central cluster (or the other way around - distributing from central cluster to regional clusters). That for example works fine as it is not active-active mirroring but just one way. So you would not need to rename the topics.

[scholzj]

Just to be clear ... I'm not trying to say that something is good or wrong idea. Just that there are many different designs and scenarios. Some are easier to do, other might not be possible with Kafka and sometimes you might have to tailor your architecture to Kafka to make it work.

[AEncinas19]

Thank you very much Jakub.