[Bug]: KafkaMirrorMaker2 Overreplication

[viscory]

Bug Description

I am using KafkaMirrorMaker2 to replicate records from one cluster to another. The replication is more or less successful for all topics (10+ topics) except 1. I ran a kafka console consumer on the topic and noticed two messages got replicated twice (20 messages in source cluster and 22 messages in target cluster.) Please provide steps on how I could diagnose this problem.

Steps to reproduce

No response

Expected behavior

No response

Strimzi version

0.37.0

Kubernetes version

1.24

Installation method

Helm

Infrastructure

Aliyun ACK

Configuration files and logs

apiVersion: [REDACTED]
items:
- apiVersion: [REDACTED].strimzi.io/[REDACTED]beta2
  kind: [REDACTED]MirrorMaker2
  metadata:
    annotations:
      kubectl.kubernetes.io/last-applied-configuration: |
        {"apiVersion":"[REDACTED].strimzi.io/[REDACTED]beta2","kind":"[REDACTED]MirrorMaker2","metadata":{"annotations":{},"labels":{"argocd.argoproj.io/instance":"strimzi-mirrormaker2-s-cn-ali"},"name":"[REDACTED]","namespace":"v2-[REDACTED]"},"spec":{"clusters":[{"alias":"[REDACTED]","bootstrapServers":"[REDACTED]-cluster-cp-[REDACTED].[REDACTED]:9092"},{"alias":"v2-[REDACTED]","bootstrapServers":"[REDACTED]-[REDACTED]-bootstrap:9092","config":{"config.storage.replication.factor":-1,"offset.storage.replication.factor":-1,"status.storage.replication.factor":-1}}],"connectCluster":"v2-[REDACTED]","jmxOptions":{},"metricsConfig":{"type":"jmxPrometheusExporter","valueFrom":{"configMapKeyRef":{"key":"metrics-config.yml","name":"mirror-maker-2-metrics"}}},"mirrors":[{"checkpointConnector":{"config":{"checkpoints.topic.replication.factor":-1,"replication.policy.class":"org.apache.[REDACTED].connect.mirror.IdentityReplicationPolicy","sync.group.offsets.enabled":"true"}},"groupsPattern":".*","sourceCluster":"[REDACTED]","sourceConnector":{"config":{"offset-syncs.topic.replication.factor":6,"replication.factor":3,"replication.policy.class":"io.strimzi.[REDACTED].connect.mirror.IdentityReplicationPolicy","replication.policy.separator":"","sync.topic.acls.enabled":"false"},"tasksMax":10},"targetCluster":"v2-[REDACTED]","topicsPattern":"[REDACTED]_[REDACTED]_.*|_schemas"}],"replicas":3,"template":{"pod":{"affinity":{"nodeAffinity":{"requiredDuringSchedulingIgnoredDuringExecution":{"nodeSelectorTerms":[{"matchExpressions":[{"key":"kubernetes.io/arch","operator":"In","values":["arm64"]}]}]}}},"metadata":{"annotations":{"prometheus.io/path":"/metrics","prometheus.io/port":"9404","prometheus.io/scrape":"true","sidecar.istio.io/inject":"false"}},"tolerations":[{"effect":"NoSchedule","key":"arch","operator":"Equal","value":"arm64"}]}},"version":"3.5.1"}}
    creationTimestamp: "2023-10-16T09:14:33Z"
    generation: 1
    labels:
      argocd.argoproj.io/instance: strimzi-mirrormaker2-s-cn-ali
    name: [REDACTED]
    namespace: v2-[REDACTED]
    resourceVersion: "390827055"
    uid: b99e7b46-f182-45b2-be7f-6e1fc0a9fb82
  spec:
    clusters:
    - alias: [REDACTED]
      bootstrapServers: [REDACTED]-cluster-cp-[REDACTED].[REDACTED]:9092
    - alias: v2-[REDACTED]
      bootstrapServers: [REDACTED]-[REDACTED]-bootstrap:9092
      config:
        config.storage.replication.factor: -1
        offset.storage.replication.factor: -1
        status.storage.replication.factor: -1
    connectCluster: v2-[REDACTED]
    jmxOptions: {}
    metricsConfig:
      type: jmxPrometheusExporter
      valueFrom:
        configMapKeyRef:
          key: metrics-config.yml
          name: mirror-maker-2-metrics
    mirrors:
    - checkpointConnector:
        config:
          checkpoints.topic.replication.factor: -1
          replication.policy.class: org.apache.[REDACTED].connect.mirror.IdentityReplicationPolicy
          sync.group.offsets.enabled: "true"
      groupsPattern: .*
      sourceCluster: [REDACTED]
      sourceConnector:
        config:
          offset-syncs.topic.replication.factor: 6
          replication.factor: 3
          replication.policy.class: io.strimzi.[REDACTED].connect.mirror.IdentityReplicationPolicy
          replication.policy.separator: ""
          sync.topic.acls.enabled: "false"
        tasksMax: 10
      targetCluster: v2-[REDACTED]
      topicsPattern: [REDACTED]_[REDACTED]_.*|_schemas
    replicas: 3
    template:
      pod:
        affinity:
          nodeAffinity:
            requiredDuringSchedulingIgnoredDuringExecution:
              nodeSelectorTerms:
              - matchExpressions:
                - key: kubernetes.io/arch
                  operator: In
                  values:
                  - arm64
        metadata:
          annotations:
            prometheus.io/path: /metrics
            prometheus.io/port: "9404"
            prometheus.io/scrape: "true"
            sidecar.istio.io/inject: "false"
        tolerations:
        - effect: NoSchedule
          key: arch
          operator: Equal
          value: arm64
    version: 3.5.1
  status:
    conditions:
    - lastTransitionTime: "2023-10-16T09:15:37.721298172Z"
      status: "True"
      type: Ready
    connectors:
    - connector:
        state: RUNNING
        worker_id: 172.22.2.16:8083
      name: [REDACTED]->v2-[REDACTED].MirrorCheckpointConnector
      tasks:
      - id: 0
        state: RUNNING
        worker_id: 172.22.2.16:8083
      type: source
    - connector:
        state: RUNNING
        worker_id: 172.22.1.240:8083
      name: [REDACTED]->v2-[REDACTED].MirrorSourceConnector
      tasks:
      - id: 0
        state: RUNNING
        worker_id: 172.22.1.240:8083
      - id: 1
        state: RUNNING
        worker_id: 172.22.2.16:8083
      - id: 2
        state: RUNNING
        worker_id: 172.22.1.31:8083
      - id: 3
        state: RUNNING
        worker_id: 172.22.1.240:8083
      - id: 4
        state: RUNNING
        worker_id: 172.22.2.16:8083
      - id: 5
        state: RUNNING
        worker_id: 172.22.1.31:8083
      - id: 6
        state: RUNNING
        worker_id: 172.22.1.240:8083
      - id: 7
        state: RUNNING
        worker_id: 172.22.2.16:8083
      - id: 8
        state: RUNNING
        worker_id: 172.22.1.31:8083
      - id: 9
        state: RUNNING
        worker_id: 172.22.1.240:8083
      type: source
    labelSelector: strimzi.io/cluster=[REDACTED],strimzi.io/name=[REDACTED]-mirrormaker2,strimzi.io/kind=[REDACTED]MirrorMaker2
    observedGeneration: 1
    replicas: 3
    url: http://[REDACTED]-mirrormaker2-api.v2-[REDACTED].svc:8083
kind: List
metadata:
  resourceVersion: ""

Additional context

No response

[scholzj]

I think the basic design of MM2 is at-least-once. So I think duplicate messages can happen. So if it is just 2 messages out of many, it could have been just some network glitch or something that caused it.

[scholzj]

Discussed on the Community call on 19.10.2023: From Apache Kafka 3.5, exactly once semantic can be enabled in the source connector. Should be converted to discussion since this looks more like an issue to ty to help with rather than Strimzi bug (Strimzi does not mirror the messages anyway, Apache Kafka does it).