You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2023/11/22 12:45:22 UTC
[PR] KAFKA-15803: Update leader epoch in commitAsync and committed [kafka]
lucasbru opened a new pull request, #14817:
URL: https://github.com/apache/kafka/pull/14817
To align the new consumer completely with the legacy consumer,
we need to update the latest seen leader epoch in the metadata
both when we commit offsets with a leader epoch using
`commitAsync` and when we fetch the latest committed offsets
using `committed`.
We add unit tests to `AsyncKafkaConsumer` to test that
metadata is correctly updated. We also add a corresponding test
for `commitSync`, which already had the leader epoch logic.
### 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15803: Update leader epoch in commitAsync and committed [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #14817:
URL: https://github.com/apache/kafka/pull/14817#discussion_r1403333866
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -204,6 +206,24 @@ public void testCommitted() {
}
}
+ @Test
+ public void testCommittedLeaderEpochUpdate() {
+ final TopicPartition t0 = new TopicPartition("t0", 2);
+ final TopicPartition t1 = new TopicPartition("t0", 3);
+ HashMap<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
+ topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L, Optional.of(2), ""));
+ topicPartitionOffsets.put(t1, null);
+
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ committedFuture.complete(topicPartitionOffsets);
+
+ try (MockedConstruction<OffsetFetchApplicationEvent> ignored = offsetFetchEventMocker(committedFuture)) {
+ assertDoesNotThrow(() -> consumer.committed(topicPartitionOffsets.keySet(), Duration.ofMillis(1000)));
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
+ verify(applicationEventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
Review Comment:
Done
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -204,6 +206,24 @@ public void testCommitted() {
}
}
+ @Test
+ public void testCommittedLeaderEpochUpdate() {
+ final TopicPartition t0 = new TopicPartition("t0", 2);
+ final TopicPartition t1 = new TopicPartition("t0", 3);
+ HashMap<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
+ topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L, Optional.of(2), ""));
+ topicPartitionOffsets.put(t1, null);
+
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ committedFuture.complete(topicPartitionOffsets);
+
+ try (MockedConstruction<OffsetFetchApplicationEvent> ignored = offsetFetchEventMocker(committedFuture)) {
+ assertDoesNotThrow(() -> consumer.committed(topicPartitionOffsets.keySet(), Duration.ofMillis(1000)));
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
Review Comment:
Done
--
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
Re: [PR] KAFKA-15803: Update leader epoch in commitAsync and committed [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #14817:
URL: https://github.com/apache/kafka/pull/14817
--
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
Re: [PR] KAFKA-15803: Update leader epoch in commitAsync and committed [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #14817:
URL: https://github.com/apache/kafka/pull/14817#issuecomment-1825403240
Test failures unrelated
--
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
Re: [PR] KAFKA-15803: Update leader epoch in commitAsync and committed [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #14817:
URL: https://github.com/apache/kafka/pull/14817#discussion_r1403241148
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -204,6 +206,24 @@ public void testCommitted() {
}
}
+ @Test
+ public void testCommittedLeaderEpochUpdate() {
+ final TopicPartition t0 = new TopicPartition("t0", 2);
+ final TopicPartition t1 = new TopicPartition("t0", 3);
+ HashMap<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
+ topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L, Optional.of(2), ""));
+ topicPartitionOffsets.put(t1, null);
+
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ committedFuture.complete(topicPartitionOffsets);
+
+ try (MockedConstruction<OffsetFetchApplicationEvent> ignored = offsetFetchEventMocker(committedFuture)) {
+ assertDoesNotThrow(() -> consumer.committed(topicPartitionOffsets.keySet(), Duration.ofMillis(1000)));
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
Review Comment:
Could you please test multiple calls to `updateLastSeenEpochIfNewer()`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -204,6 +206,24 @@ public void testCommitted() {
}
}
+ @Test
+ public void testCommittedLeaderEpochUpdate() {
+ final TopicPartition t0 = new TopicPartition("t0", 2);
+ final TopicPartition t1 = new TopicPartition("t0", 3);
+ HashMap<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
+ topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L, Optional.of(2), ""));
+ topicPartitionOffsets.put(t1, null);
+
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ committedFuture.complete(topicPartitionOffsets);
+
+ try (MockedConstruction<OffsetFetchApplicationEvent> ignored = offsetFetchEventMocker(committedFuture)) {
+ assertDoesNotThrow(() -> consumer.committed(topicPartitionOffsets.keySet(), Duration.ofMillis(1000)));
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
+ verify(applicationEventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
Review Comment:
Is there a reason you put these two verifications inside the try-resource-block but not in the other tests?
--
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