You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2023/11/24 09:47:38 UTC
(kafka) branch trunk updated: KAFKA-15803: Update leader epoch in commitAsync and committed (#14817)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 768a458b126 KAFKA-15803: Update leader epoch in commitAsync and committed (#14817)
768a458b126 is described below
commit 768a458b126c81a57e654b67befe7eb59beb2e17
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Fri Nov 24 10:47:31 2023 +0100
KAFKA-15803: Update leader epoch in commitAsync and committed (#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.
Reviewers: Andrew Schofield <as...@confluent.io>, Bruno Cadonna <ca...@apache.org>
---
.../consumer/internals/AsyncKafkaConsumer.java | 9 ++-
.../consumer/internals/AsyncKafkaConsumerTest.java | 81 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 2 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index b7e61ec07ba..6c55734b4ea 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -512,6 +512,9 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
maybeThrowFencedInstanceException();
maybeThrowInvalidGroupIdException();
+ log.debug("Committing offsets: {}", offsets);
+ offsets.forEach(this::updateLastSeenEpochIfNewer);
+
if (offsets.isEmpty()) {
return CompletableFuture.completedFuture(null);
}
@@ -656,7 +659,10 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
final OffsetFetchApplicationEvent event = new OffsetFetchApplicationEvent(partitions);
wakeupTrigger.setActiveTask(event.future());
try {
- return applicationEventHandler.addAndGet(event, time.timer(timeout));
+ final Map<TopicPartition, OffsetAndMetadata> committedOffsets = applicationEventHandler.addAndGet(event,
+ time.timer(timeout));
+ committedOffsets.forEach(this::updateLastSeenEpochIfNewer);
+ return committedOffsets;
} finally {
wakeupTrigger.clearTask();
}
@@ -941,7 +947,6 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
long commitStart = time.nanoseconds();
try {
CompletableFuture<Void> commitFuture = commit(offsets, true);
- offsets.forEach(this::updateLastSeenEpochIfNewer);
ConsumerUtils.getResult(commitFuture, time.timer(timeout));
} finally {
wakeupTrigger.clearTask();
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
index 705eaca3140..55afdbfe080 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import java.util.Arrays;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
@@ -24,6 +25,7 @@ import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
import org.apache.kafka.clients.consumer.internals.events.AssignmentChangeApplicationEvent;
+import org.apache.kafka.clients.consumer.internals.events.CommitApplicationEvent;
import org.apache.kafka.clients.consumer.internals.events.ListOffsetsApplicationEvent;
import org.apache.kafka.clients.consumer.internals.events.NewTopicsMetadataUpdateRequestEvent;
import org.apache.kafka.clients.consumer.internals.events.OffsetFetchApplicationEvent;
@@ -212,6 +214,27 @@ public class AsyncKafkaConsumerTest {
}
}
+ @Test
+ public void testCommittedLeaderEpochUpdate() {
+ final TopicPartition t0 = new TopicPartition("t0", 2);
+ final TopicPartition t1 = new TopicPartition("t0", 3);
+ final TopicPartition t2 = new TopicPartition("t0", 4);
+ HashMap<TopicPartition, OffsetAndMetadata> topicPartitionOffsets = new HashMap<>();
+ topicPartitionOffsets.put(t0, new OffsetAndMetadata(10L, Optional.of(2), ""));
+ topicPartitionOffsets.put(t1, null);
+ topicPartitionOffsets.put(t2, new OffsetAndMetadata(20L, Optional.of(3), ""));
+
+ 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(testBuilder.metadata).updateLastSeenEpochIfNewer(t2, 3);
+ verify(applicationEventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ }
+
@Test
public void testCommitted_ExceptionThrown() {
Map<TopicPartition, OffsetAndMetadata> offsets = mockTopicPartitionOffset();
@@ -346,6 +369,49 @@ public class AsyncKafkaConsumerTest {
"This method is deprecated and will be removed in the next major release.", e.getMessage());
}
+ @Test
+ public void testCommitSyncLeaderEpochUpdate() {
+ 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, new OffsetAndMetadata(20L, Optional.of(1), ""));
+
+ consumer.assign(Arrays.asList(t0, t1));
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ commitFuture.complete(null);
+
+ try (MockedConstruction<CommitApplicationEvent> ignored = commitEventMocker(commitFuture)) {
+ assertDoesNotThrow(() -> consumer.commitSync(topicPartitionOffsets));
+ }
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t1, 1);
+ verify(applicationEventHandler).add(ArgumentMatchers.isA(CommitApplicationEvent.class));
+ }
+
+ @Test
+ public void testCommitAsyncLeaderEpochUpdate() {
+ MockCommitCallback callback = new MockCommitCallback();
+ 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, new OffsetAndMetadata(20L, Optional.of(1), ""));
+
+ consumer.assign(Arrays.asList(t0, t1));
+
+ CompletableFuture<Void> commitFuture = new CompletableFuture<>();
+ commitFuture.complete(null);
+
+ try (MockedConstruction<CommitApplicationEvent> ignored = commitEventMocker(commitFuture)) {
+ assertDoesNotThrow(() -> consumer.commitAsync(topicPartitionOffsets, callback));
+ }
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t0, 2);
+ verify(testBuilder.metadata).updateLastSeenEpochIfNewer(t1, 1);
+ verify(applicationEventHandler).add(ArgumentMatchers.isA(CommitApplicationEvent.class));
+ }
+
@Test
public void testEnsurePollExecutedCommitAsyncCallbacks() {
MockCommitCallback callback = new MockCommitCallback();
@@ -427,6 +493,21 @@ public class AsyncKafkaConsumerTest {
return mockConstruction(OffsetFetchApplicationEvent.class, mockInitializer);
}
+ private static MockedConstruction<CommitApplicationEvent> commitEventMocker(CompletableFuture<Void> future) {
+ Answer<Void> getInvocationAnswer = invocation -> {
+ Timer timer = invocation.getArgument(0);
+ return ConsumerUtils.getResult(future, timer);
+ };
+
+ MockedConstruction.MockInitializer<CommitApplicationEvent> mockInitializer = (mock, ctx) -> {
+ when(mock.get(any())).thenAnswer(getInvocationAnswer);
+ when(mock.type()).thenReturn(ApplicationEvent.Type.COMMIT);
+ when(mock.future()).thenReturn(future);
+ };
+
+ return mockConstruction(CommitApplicationEvent.class, mockInitializer);
+ }
+
@Test
public void testAssign() {
final TopicPartition tp = new TopicPartition("foo", 3);