Some consumer groups are missing during translation using mirror maker 2

[gurufw20]

We are working on a migration of kafka. We are actually moving kafka from ec2 to strimzi using mirror maker 2. I don't see any issue while translation the topics and its events. however, when it comes to consumer groups, not all the consumer groups are getting translated eventhough all are actively consuming. I had like 4000 consumer groups but only 2104 consumer groups got translated. It's ok for me if there is some lag during the translation but we can't have completely missing consumer groups. Here is my mirrormaker2 configuration. We are using kafka 3.7.1 and strimzi version 0.43.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
 name: mirror-maker-2
 annotations:
   strimzi.io/use-prometheus: "true"
spec:
 replicas: 40
 version: 3.7.1  
 connectCluster: "target-a"
 metricsConfig:
   type: jmxPrometheusExporter
   valueFrom:
     configMapKeyRef:
       name: mirrormaker-metrics
       key: metrics-config.yml
 clusters:
   - alias: "source-a"
     bootstrapServers: kafka-source.kafka:9094
     authentication:
       type: plain
       username: NDSNQ6GM89NAB3MG
       passwordSecret:
        secretName: my-ec2-secret
        password: plain.password
     config:
       security.protocol: SASL_PLAINTEXT
       sasl.mechanism: PLAIN

   - alias: "target-a"
     bootstrapServers: kafka-target.kafka:9094
     authentication:
       type: scram-sha-512
       username: my-user
       passwordSecret:
        secretName: my-kafka-user-secret-new
        password: plain.password
     config:
       rebalance.timeout.ms: 30000
       connections.max.idle.ms: 600000
       offset.flush.interval.ms: 30000
       producer.batch.size: 655360
       producer.acks: all
       producer.linger.ms: 10
       producer.buffer.memory: 1073741824
       producer.compression.type: none
       security.protocol: SASL_PLAINTEXT 
       sasl.mechanism: SCRAM-SHA-512
       producer.max.request.size: 10485760
       offset_flush_timeout: 250000
       offset.storage.replication.factor: -1
       status.storage.replication.factor: -1
       config.storage.replication.factor: -1
       status.storage.topic: mm2-status.source-a.internal
       config.storage.topic: mm2-configs.source-a.internal

 mirrors:
   - sourceCluster: "source-a"
     targetCluster: "target-a"
     sourceConnector:
       tasksMax: 500
       autoRestart:
        enabled: true
       config:
         producer.override.batch.size: 655360
         producer.override.linger.ms: 100
         producer.request.timeout.ms: 60000
         admin.timeout.ms: 60000
         connector.client.config.override.policy: All
         exactly.once.source.support: disabled
         consumer.isolation.level: read_committed
         replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
         errors.retry.timeout: 500
         errors.retry.delay.max.ms: 100
         errors.log.enable: true
         offset-syncs.topic.location: "target"
         offset.lag.max: 100 
         refresh.topics.enabled: true
         sync.topic.acls.enabled: true
         sync.topic.configs.enabled: true
         replication.factor: 3
         offset-syncs.topic.replication.factor: 3 
         refresh.topics.interval.seconds: 60  
         refresh.topic.list: true
         consumer.poll.timeout.ms: 10000
         consumer.max.poll.records: 2000
         consumer.fetch.max.wait.ms: 100
         consumer.fetch.max.bytes: 104857600 
         consumer.max.poll.interval.ms: 600000
         consumer.heartbeat.interval.ms: 6000
         consumer.session.timeout.ms: 60000
         consumer.fetch.min.bytes: 1048576
         consumer.max.partition.fetch.bytes: 52428800 
         offset.syncs.enable: true
         consumer.request.timeout.ms: 60000
         offset.syncs.commit.interval.ms: 60000
         emit.offset.syncs.enabled: true
         transaction.timeout.ms: 900000     
         max.poll.interval.ms: 1800000      
         enable.idempotence: true           
         consumer.auto.offset.reset: "latest"
         
     checkpointConnector:
       tasksMax: 100
       autoRestart:
        enabled: true
       config:
         admin.timeout.ms: 120000         
         replication.policy.class: "org.apache.kafka.connect.mirror.IdentityReplicationPolicy"
         emit.checkpoints.enabled: true
         emit.checkpoints.interval.seconds: 5
         sync.group.offsets.enabled: true
         sync.group.offsets.interval.seconds: 5
         offset-syncs.topic.location: "target"  
         refresh.groups.interval.seconds: 10
         consumer.poll.timeout.ms: 200
         checkpoints.topic.replication.factor: 3
         consumer.max.poll.records: 500
         consumer.session.timeout.ms: 30000 
         offset-syncs.topic.replication.factor: 3  
         groups: .*
         
     heartbeatConnector: 
      autoRestart:
        enabled: true
      config:
        heartbeats.topic.replication.factor: 3
        emit.heartbeats.enabled: true
        
     topicsPattern: ".*"
     groupsPattern: ".*"
     
     
 resources:
   requests:
     memory: "12Gi"
     cpu: "2"
   limits:
     memory: "12Gi"
     cpu: "2"
 template:
   pod:
    metadata:
    securityContext:
       runAsGroup: 65534
       runAsNonRoot: true
       runAsUser: 65534
   connectContainer:
     securityContext:
       readOnlyRootFilesystem: true
       allowPrivilegeEscalation: false
     env:
       - name: KAFKA_HEAP_OPTS
         value: "-Xms6g -Xmx6g"
       - name: GC_LOG_ENABLED
         value: "true"
       - name: KAFKA_JVM_PERFORMANCE_OPTS
         value: "-XX:+UseG1GC -XX:MaxGCPauseMillis=100 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -XX:G1HeapRegionSize=16M"
       - name: STRIMZI_JAVA_SYSTEM_PROPERTIES
         value: "-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dkafka.logs.dir=/opt/kafka -Dlog4j.configuration=file:/opt/kafka/custom-config/log4j.properties"

 readinessProbe:
   initialDelaySeconds: 180
   timeoutSeconds: 15
 livenessProbe:
   initialDelaySeconds: 180
   timeoutSeconds: 15

[katheris]

Hey @gurufw20 I've been taking a look at this, can I ask what method you are using to verify that the consumer groups got translated? Are you finding that the consumers are starting from the beginning of the topic again when connecting to the target cluster?

[gurufw20]

Hi @katheris Thanks for your reply. I used the below command in both source and target cluster. I could see the mismatch in count.

kafka_v1/kafka_2.13-3.7.1/bin/kafka-consumer-groups.sh --bootstrap-server kafka-1.kafka:9094 --command-config kafka_v1/kafka_2.13-3.7.1/config/config.properties --list | wc -l

[katheris]

Thanks @gurufw20, I'm not certain whether the consumer groups will automatically get listed in the target cluster, however the consumer offsets should have been stored in the <TOPIC_NAME>.checkpoints.internal topic. If you run a command similar to the following against the target cluster do you see offsets for all of your consumer groups?

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic my-topic.checkpoints.internal \
  --formatter org.apache.kafka.connect.mirror.formatters.CheckpointFormatter \
  --from-beginning