aws-samples / aws-samples/mirrormaker2-msk-migration

Consumer offsets are not getting translated to destination

Open
#2 5 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
HCL
Stars
34
Forks
37
PR merge metrics
No merged PRs in 30d

Description

We have run the mirrormaker2 on a kafka connect framework as per the instructions and notice the topics/data migrated from source to destination.
However the offsets on the destination seem to be started from "0" and not translated from the source. This would leave all clients to consume the offsets from the beginning after switching to MSK. I am assuming some issue with the consumer group configuration but not able to identify what is missing.

Below are the configurations used:
mm2-msc-cust-repl-policy:
{
"name": "mm2-msc",
"connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
"clusters": "msksource,mskdest",
"source.cluster.alias": "msksource",
"target.cluster.alias": "mskdest",
"target.cluster.bootstrap.servers": "Destination MSK bootstraps",
"source.cluster.bootstrap.servers": "Source kafka bootstraps",
"topics": ".*",
"tasks.max": "4",
"key.converter": " org.apache.kafka.connect.converters.ByteArrayConverter",
"value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
"replication.policy.class": "com.amazonaws.kafka.samples.CustomMM2ReplicationPolicy",
"replication.factor": "3",
"offset-syncs.topic.replication.factor": "3",
"sync.topic.configs.interval.seconds": "20",
"refresh.topics.interval.seconds": "20",
"refresh.groups.interval.seconds": "20",
"consumer.group.id": "preprod-cgi",
"producer.enable.idempotence":"true",
"sync.topic.acls.enabled": "false"
}
The consumer.group.id specified here was not created by mirrormaker2, we used the same consumer.group.id while running the Group offset sync application and it could not identify the consumer group. Our understanding is this is the consumer group created for mirror maker consumer and after the mirror maker producer writes the data to MSK these offsets with respect the "preprod-cgi" will be translated in destination.

java -jar /tmp/kafka/MM2GroupOffsetSync-1.0-SNAPSHOT.jar -cgi preprod-cgi -src msksource-ps -pfp /tmp/kafka/consumer.properties -rpc com.amazonaws.kafka.samples.CustomMM2ReplicationPolicy 2>&1 > /tmp/preprod-v4.log &

mm2-cpc-cust-repl-policy:
{

"target.cluster.bootstrap.servers": "Destination MSK bootstraps",
"source.cluster.bootstrap.servers": "Source kafka bootstraps",
"tasks.max": "1",
"key.converter": " org.apache.kafka.connect.converters.ByteArrayConverter",
"value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
"replication.policy.class": "com.amazonaws.kafka.samples.CustomMM2ReplicationPolicy",
"replication.factor": "3",
"checkpoints.topic.replication.factor": "3",
"emit.checkpoints.interval.seconds": "20"

}

Also while starting the kafka connect service, it uses a connect-distributed.properties file which has a group.id in it. Does this have any particular significance or can create any conflicts with the above consumer group id from mm2-msc-cust-repl-policy:??

Reference for source vs destination offsets post migration:
Source:
agent-payloads:0:235943806
agent-payloads:1:0
agent-payloads:2:0
agent-payloads:3:0
agent-payloads:4:0

Destination:
agent-payloads:0:24483600
agent-payloads:1:0
agent-payloads:2:0
agent-payloads:3:0
agent-payloads:4:0

Thanks and appreciate any help on this.

Thanks,
Shravan

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start by reviewing the MirrorSourceConnector and checkpoint connector configurations, then inspect the MM2GroupOffsetSync command and the group.id in connect-distributed.properties. Compare the source and destination offsets for the listed topic and determine what configuration or synchronization step is needed for translated destination offsets.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.