You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/05/07 03:02:31 UTC
[GitHub] [kafka] ning2008wisc opened a new pull request #10644: [KAFKA-12635] MM 2.0: auto sync consumer offset 0 when translated off…
ning2008wisc opened a new pull request #10644:
URL: https://github.com/apache/kafka/pull/10644
…set larger than partition end offset
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Dragotic commented on pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
Dragotic commented on pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#issuecomment-860509704
Hello team!
@ning2008wisc thank you very much for the bug fix, any update on when we can expect this to be merged to the upstream and released?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] tyamashi-oss commented on pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
tyamashi-oss commented on pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#issuecomment-1006638780
I'm having a similar problem. I was looking for a solution to my problem and found here. Please forgive me for expressing my personal opinion and sharing my similar problem.
I believe the consumer offset should be log-end-offset, not 0, when translated offset larger than partition end offset.
(Actually, in the original problem reproduction, log-end-offset is also 0, so the current implementation that resets offset to 0 will be able to be simply replaced with log-end-offset.)
When translated offset larger than partition end offset or lower than partition earliest offset, once the consumer client start in the target cluster, the consumer client will be reset based on "auto.offset.reset(default: latest)" because the broker will return InvalidOffsetException. Then the consumer in the target cluster will duplicate or skip a large number of messages.
In order to avoid automatic reset by the consumer, I believe that the translated offset should stay within the valid log offset(from earliest offset to end offset). So the consumer offset should be log-end-offset, not 0, when translated offset larger than partition end offset.
FYI, in addition to the original problem, I also found another way to reproduce the negative lag caused by the translated offset being larger than the partition end offset. In the following case, the solution is to use log-end-offset(, not 0 because it will lead a large number of duplicate messages) for the translated offset.
~~~
0. The topic can hold only 100,000 messages based on retention.bytes.
1. Add 10,000 messages to the topic on the source cluster and consume it by a consumer on the source cluster, then wait for MM2 to synchronize.
=>
The source cluster:
log: from 1 to 10,000
Consumer group offset : 10,000
lag: 0
The target cluster:
log: from 1 to 10,000
Consumer group offset : 10,000
lag: 0
2. Stop MM2 and add 1,000,000 messages to the topic on the source cluster and consume it by a consumer on the source cluster.
=>
The source cluster:
log: from 910,001 to 1,010,000 (cut by retention.bytes)
Consumer group offset : 1,010,000
lag: 0
The target cluster [not changed]:
log: from 1 to 10,000
Consumer group offset : 10,000
lag: 0
3. Restart MM2, then wait for MM2 to synchronize.
Although this is a timing issue, the problem occurs if the MirrorCheckpointConnector synchronizes the consumer group offset before the MirrorSourceConnector updates the offsetSyncTopic.
=>
The source cluster [not changed]:
log: from 910,001 to 1,010,000
Consumer group offset : 1,010,000
lag: 0
The target cluster:
log: from 10,001 to 110,000
Consumer group offset : 1,010,000
lag: -900,000
~~~
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#discussion_r794275490
##########
File path: connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java
##########
@@ -243,6 +251,45 @@ private void refreshIdleConsumerGroupOffset() {
}
}
}
+
+ private void refreshTopicPartitionEndOffset() {
+
+ Map<TopicPartition, Long> topicPartitionOffsets = new HashMap<>();
+ for (Entry<String, Map<TopicPartition, OffsetAndMetadata>> group : getConvertedUpstreamOffset().entrySet()) {
+ Map<TopicPartition, OffsetAndMetadata> convertedUpstreamOffset = group.getValue();
+ for (Entry<TopicPartition, OffsetAndMetadata> offset : convertedUpstreamOffset.entrySet()) {
+ topicPartitionOffsets.put(offset.getKey(), offset.getValue().offset());
+ }
+ }
+ if (topicPartitionOffsets.keySet().size() != 0) {
+ KafkaConsumer<byte[], byte[]> consumer = MirrorUtils.newConsumer(config.targetConsumerConfig());
+ consumer.assign(topicPartitionOffsets.keySet());
+ topicPartitionOffsets.forEach(consumer::seek);
+ consumer.poll(Duration.ofMillis(500));
+ topicPartitionEndOffset = consumer.endOffsets(topicPartitionOffsets.keySet());
+ consumer.close();
+ }
Review comment:
`targetAdminClient.listOffsets()` is easier and more efficient to get the LEO.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ning2008wisc commented on pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
ning2008wisc commented on pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#issuecomment-841893030
users have verified that the PR is working. Please take the initial review when available cc @ryannedolan @mimaison Thanks
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] scanterog commented on pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
scanterog commented on pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#issuecomment-1001606207
Hello! What's the current status of this PR?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ning2008wisc commented on pull request #10644: [KAFKA-12635] auto sync consumer offset 0 when translated offset larger than partition end offset
Posted by GitBox <gi...@apache.org>.
ning2008wisc commented on pull request #10644:
URL: https://github.com/apache/kafka/pull/10644#issuecomment-841893030
users have verified that the PR is working. Please take the initial review when available cc @ryannedolan @mimaison Thanks
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org