[KAFKA-17186] Cannot receive message after stopping Source Mirror Maker 2

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 3.7.1
• Component/s: mirrormaker
• Environment: Source Kafka Cluster per Node: ; CPU(s): 32 ; Memory: 32G/1.1G free ; ; Target Kafka Cluster standalone Node: ; CPU(s): 24 ; Memory: 30G/909M free ; ; Kafka Version 3.7 ; Mirrormaker Version 3.7.1
• Assignee: Unassigned
• Reporter: George Yang
• Created: 23/Jul/24 10:55
• Updated: 30/Jul/24 01:07
• Resolved: 30/Jul/24 01:07

Description

Deploy nodes 1, 2, and 3 in Data Center A, with MM2 service deployed on node 1. Deploy node 1 in Data Center B, with MM2 service also deployed on node 1. Currently, a service on node 1 in Data Center A acts as a producer sending messages to the `myTest` topic. A service in Data Center B acts as a consumer listening to `A.myTest`. 

The issue arises when MM2 on node 1 in Data Center A is stopped: the consumer in Data Center B ceases to receive messages. Even after I restarting MM2 in Data Center A, the consumer in Data Center B still does not receive messages until approximately 5 minutes later when a rebalance occurs, at which point it begins receiving messages again.

 

[Logs From Consumer on Data Center B]

[2024-07-23 17:29:17,270] INFO [MirrorCheckpointConnector|worker] refreshing consumer groups took 185 ms (org.apache.kafka.connect.mirror.Scheduler:95)
[2024-07-23 17:29:19,189] INFO [MirrorCheckpointConnector|worker] refreshing consumer groups took 365 ms (org.apache.kafka.connect.mirror.Scheduler:95)
[2024-07-23 17:29:22,271] INFO [MirrorCheckpointConnector|worker] refreshing consumer groups took 186 ms (org.apache.kafka.connect.mirror.Scheduler:95)
[2024-07-23 17:29:24,193] INFO [MirrorCheckpointConnector|worker] refreshing consumer groups took 369 ms (org.apache.kafka.connect.mirror.Scheduler:95)
[2024-07-23 17:29:25,377] INFO [Worker clientId=B->A, groupId=B-mm2] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:242)
[2024-07-23 17:29:25,377] INFO [Worker clientId=B->A, groupId=B-mm2] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:604)
[2024-07-23 17:29:25,386] INFO [Worker clientId=B->A, groupId=B-mm2] Successfully joined group with generation Generation{generationId=52, memberId='B->A-adc19038-a8b6-40fb-9bf6-249f866944ab', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:665)
[2024-07-23 17:29:25,390] INFO [Worker clientId=B->A, groupId=B-mm2] Successfully synced group in generation Generation{generationId=52, memberId='B->A-adc19038-a8b6-40fb-9bf6-249f866944ab', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:842)
[2024-07-23 17:29:25,390] INFO [Worker clientId=B->A, groupId=B-mm2] Joined group at generation 52 with protocol version 2 and got assignment: Assignment{error=0, leader='B->A-adc19038-a8b6-40fb-9bf6-249f866944ab', leaderUrl='NOTUSED', offset=1360, connectorIds=[MirrorCheckpointConnector], taskIds=[MirrorCheckpointConnector-0, MirrorCheckpointConnector-1, MirrorCheckpointConnector-2], revokedConnectorIds=[], revokedTaskIds=[], delay=0} with rebalance delay: 0 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:2580)
[2024-07-23 17:29:25,390] INFO [Worker clientId=B->A, groupId=B-mm2] Starting connectors and tasks using config offset 1360 (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1921)
[2024-07-23 17:29:25,390] INFO [Worker clientId=B->A, groupId=B-mm2] Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:1950)
[2024-07-23 17:29:26,883] INFO [Worker clientId=A->B, groupId=A-mm2] Rebalance started (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:242)
[2024-07-23 17:29:26,883] INFO [Worker clientId=A->B, groupId=A-mm2] (Re-)joining group (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:604)
[2024-07-23 17:29:26,890] INFO [Worker clientId=A->B, groupId=A-mm2] Successfully joined group with generation Generation{generationId=143, memberId='A->B-0d04e6c1-f12a-4121-89af-e9992a167a01', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:665)
[2024-07-23 17:29:26,893] INFO [Worker clientId=A->B, groupId=A-mm2] Successfully synced group in generation Generation{generationId=143, memberId='A->B-0d04e6c1-f12a-4121-89af-e9992a167a01', protocol='sessioned'} (org.apache.kafka.connect.runtime.distributed.WorkerCoordinator:842)

[Configuration]

name=MCS-MM2
clusters = A, B

A.bootstrap.servers = [kafka1]:[port],[kafka2]:[port],[kafka3]:[port]
B.bootstrap.servers = [kafka]:[port]

1. enable and configure individual replication flows
   A->B.enabled = true
   A->B.topics = .*

B->A.enabled = true
B->A.topics = .*

replication.factor=2

tasks.max=3
emit.checkpoints.interval.seconds=5
A.producer.acks=all
A.producer.batch.size=50000

A.consumer.auto.offset.reset=latest
B.consumer.auto.offset.reset=latest
A.consumer.enable.auto.commit=true
B.consumer.enable.auto.commit=true
A.consumer.max.poll.interval.ms=20000
B.consumer.max.poll.interval.ms=20000

checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1

offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1

refresh.topics.enabled=true
refresh.topics.interval.seconds=5

refresh.groups.enabled=true
refresh.groups.interval.seconds=5