lucasbru commented on code in PR #15613:
URL: https://github.com/apache/kafka/pull/15613#discussion_r1568770827
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1388,6 +1393,31 @@ public void commitSync(Map<TopicPartition,
OffsetAndMetadata> offsets, Duration
}
}
+ private void awaitPendingAsyncCommitsAndExecuteCommitCallbacks(Timer
timer, boolean disableWakeup) {
+ if (lastPendingAsyncCommit == null) {
+ return;
+ }
+
+ try {
+ final CompletableFuture<Void> futureToAwait = new
CompletableFuture<>();
+ // We don't want the wake-up trigger to complete our pending async
commit future,
+ // so create new future here. Any errors in the pending async
commit will be handled
+ // by the async commit future / the commit callback - here, we
just want to wait for it to complete.
+ lastPendingAsyncCommit.whenComplete((v, t) ->
futureToAwait.complete(null));
+ if (!disableWakeup) {
+ wakeupTrigger.setActiveTask(futureToAwait);
+ }
+ ConsumerUtils.getResult(futureToAwait, timer);
Review Comment:
Yes. I think if `lastPendingAsyncCommit` is completed before entering here,
the `whenComplete` will execute immediately.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -158,7 +158,11 @@ public class AsyncKafkaConsumerTest {
public void resetAll() {
backgroundEventQueue.clear();
if (consumer != null) {
- consumer.close(Duration.ZERO);
+ try {
+ consumer.close(Duration.ZERO);
+ } catch (Exception e) {
+ assertInstanceOf(KafkaException.class, e);
+ }
Review Comment:
`resetAll` isn't supposed to test anything, so this also shouldn't mask
anything. It's purely for cleanup. In this case, it only affects two tests that
will timeout on close (since we don't mock an async commit response). So let me
do it. But in general, I wonder if adding clean-up logic to the tests itself
won't reduce readability/clarity of the actual test case.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -616,6 +620,80 @@ public void
testCommitSyncTriggersFencedExceptionFromCommitAsync() {
assertEquals("Get fenced exception for group.instance.id
groupInstanceId1", e.getMessage());
}
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithEmptyOffsets() {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
+ consumer.commitAsync();
+
+ // Commit async is not completed yet, so commit sync should wait for
it to complete (time out)
+ assertThrows(TimeoutException.class, () ->
consumer.commitSync(Collections.emptyMap(), Duration.ofMillis(100)));
+
+ // Complete async commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
Review Comment:
Yes. I agree, it's a bit weird, but mockito is recording those invocations
with copies (references) of all arguments. That's also why spying on lots of
objects in busy event loops will accumulate lots of such "invocation objects"
in memory.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -616,6 +620,80 @@ public void
testCommitSyncTriggersFencedExceptionFromCommitAsync() {
assertEquals("Get fenced exception for group.instance.id
groupInstanceId1", e.getMessage());
}
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithEmptyOffsets() {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
+ consumer.commitAsync();
Review Comment:
I guess I may be less concerned with code duplication in test setup than my
reviewers :). Done. Added another helper method that contains the lines up to
the incomplete async commit.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -616,6 +620,80 @@ public void
testCommitSyncTriggersFencedExceptionFromCommitAsync() {
assertEquals("Get fenced exception for group.instance.id
groupInstanceId1", e.getMessage());
}
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithEmptyOffsets() {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
+ consumer.commitAsync();
+
+ // Commit async is not completed yet, so commit sync should wait for
it to complete (time out)
+ assertThrows(TimeoutException.class, () ->
consumer.commitSync(Collections.emptyMap(), Duration.ofMillis(100)));
+
+ // Complete async commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
+ final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
+ commitEvent.future().complete(null);
+
+ // Commit async is completed, so commit sync completes immediately
(since offsets are empty)
+ assertDoesNotThrow(() -> consumer.commitSync(Collections.emptyMap(),
Duration.ofMillis(100)));
+ }
+
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithNonEmptyOffsets()
{
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ testSyncCommitTimesoutAfterIncompleteAsyncCommit(tp);
+
+ // Complete async commit event and sync commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
+ final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
+ commitEvent.future().complete(null);
+
+ // Commit async is completed, so commit sync completes immediately
(since offsets are empty)
+ assertDoesNotThrow(() ->
consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(20)),
Duration.ofMillis(100)));
+ }
+
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncButDoesNotFail() {
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ testSyncCommitTimesoutAfterIncompleteAsyncCommit(tp);
+
+ // Complete exceptionally async commit event and sync commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
+ final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
+ commitEvent.future().completeExceptionally(new KafkaException("Test
exception"));
+
+ // Commit async is completed exceptionally, but this will be handled
by commit callback - commit sync should not fail.
+ assertDoesNotThrow(() ->
consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(20)),
Duration.ofMillis(100)));
+ }
+
+ private void
testSyncCommitTimesoutAfterIncompleteAsyncCommit(TopicPartition tp) {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
Review Comment:
It's supposed to give the consumer a valid position, otherwise the commit
async doesn't do anything. So yeah, I think that would mean forcing out of
`INITIALIZING`? Or what did you mean by `INITIALIZED`..
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1388,6 +1393,31 @@ public void commitSync(Map<TopicPartition,
OffsetAndMetadata> offsets, Duration
}
}
+ private void awaitPendingAsyncCommitsAndExecuteCommitCallbacks(Timer
timer, boolean disableWakeup) {
Review Comment:
Done
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -1005,6 +1083,43 @@ public void testNoWakeupInCloseCommit() {
assertFalse(capturedEvent.get().future().isCompletedExceptionally());
}
+ @Test
+ public void testCloseAwaitPendingAsyncCommitIncomplete() {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
+
+ consumer.commitAsync();
+ Exception e = assertThrows(KafkaException.class, () ->
consumer.close(Duration.ofMillis(10)));
+ assertInstanceOf(TimeoutException.class, e.getCause());
Review Comment:
True. In this case, we are actually testing `close`. But there are other
test cases, which time-out similarly, but that's only an artifact and not the
goal of the test.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -793,9 +795,9 @@ public void commitAsync(Map<TopicPartition,
OffsetAndMetadata> offsets, OffsetCo
}
private CompletableFuture<Void> commit(final CommitEvent commitEvent) {
- maybeThrowFencedInstanceException();
- maybeInvokeCommitCallbacks();
maybeThrowInvalidGroupIdException();
+ maybeThrowFencedInstanceException();
+ offsetCommitCallbackInvoker.executeCallbacks();
Review Comment:
Not needed for correctness. I just noticed that the order in the legacy
consumer is this way. Not sure if there is a strong reason to have this or the
other order, so I decided to be consistent with the old behavior.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -616,6 +620,80 @@ public void
testCommitSyncTriggersFencedExceptionFromCommitAsync() {
assertEquals("Get fenced exception for group.instance.id
groupInstanceId1", e.getMessage());
}
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithEmptyOffsets() {
+ time = new MockTime(1);
+ consumer = newConsumer();
+
+ // Commit async (incomplete)
+
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
+
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ consumer.assign(Collections.singleton(tp));
+ consumer.seek(tp, 20);
+ consumer.commitAsync();
+
+ // Commit async is not completed yet, so commit sync should wait for
it to complete (time out)
+ assertThrows(TimeoutException.class, () ->
consumer.commitSync(Collections.emptyMap(), Duration.ofMillis(100)));
+
+ // Complete async commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
+ final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
+ commitEvent.future().complete(null);
+
+ // Commit async is completed, so commit sync completes immediately
(since offsets are empty)
+ assertDoesNotThrow(() -> consumer.commitSync(Collections.emptyMap(),
Duration.ofMillis(100)));
+ }
+
+ @Test
+ public void testCommitSyncAwaitsCommitAsyncCompletionWithNonEmptyOffsets()
{
+ final TopicPartition tp = new TopicPartition("foo", 0);
+ testSyncCommitTimesoutAfterIncompleteAsyncCommit(tp);
+
+ // Complete async commit event and sync commit event
+ final ArgumentCaptor<AsyncCommitEvent> commitEventCaptor =
ArgumentCaptor.forClass(AsyncCommitEvent.class);
+ verify(applicationEventHandler).add(commitEventCaptor.capture());
+ final AsyncCommitEvent commitEvent = commitEventCaptor.getValue();
+ commitEvent.future().complete(null);
+
+ // Commit async is completed, so commit sync completes immediately
(since offsets are empty)
Review Comment:
Good catch!
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
