aws-samples / aws-samples/mirrormaker2-msk-migration
Consumer offsets are not getting translated to destination
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
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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