edoardocomar commented on PR #15910:
URL: https://github.com/apache/kafka/pull/15910#issuecomment-2110417331
testing results in the following scenario:
- produce 10000 records
- consume them 250 at a time
- stop mm2
- produce 10000 records
- restart consuming
- restart mm2
Emitted Checkpoints:
```
% bin/kafka-console-consumer.sh --bootstrap-server localhost:9192 \
--topic source.checkpoints.internal \
--formatter org.apache.kafka.connect.mirror.formatters.CheckpointFormatter
\
--from-beginning
```
NEW implementation with checkpoints read by Checkpoint task
```
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=500, downstreamOffset=1, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=6250, downstreamOffset=5820, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=7250, downstreamOffset=7228, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=8750, downstreamOffset=8658, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=9500, downstreamOffset=9362, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=9750, downstreamOffset=9714, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=10000, downstreamOffset=10000, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=12250, downstreamOffset=11519, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=14500, downstreamOffset=14335, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=17500, downstreamOffset=17195, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=18750, downstreamOffset=18603, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19500, downstreamOffset=19307, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19750, downstreamOffset=19659, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=20000, downstreamOffset=20000, metadata=}
```
NEW implementation with checkpoints FAILED read by Checkpoint task
```
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=10000, downstreamOffset=10000, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=11250, downstreamOffset=10000, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=16000, downstreamOffset=15919, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=17500, downstreamOffset=17349, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19000, downstreamOffset=18757, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19500, downstreamOffset=19461, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=20000, downstreamOffset=20000, metadata=}
```
Original implementation (prior to this PR)
```
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=10000, downstreamOffset=10000, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=11250, downstreamOffset=10000, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=16250, downstreamOffset=16249, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=18000, downstreamOffset=17987, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19000, downstreamOffset=18768, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=19750, downstreamOffset=19747, metadata=}
Checkpoint{consumerGroupId=mygroup1, topicPartition=source.mytopic-0,
upstreamOffset=20000, downstreamOffset=20000, metadata=}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
