KAFKA-17480: New consumer commit all consumed should retrieve offsets in background thread #17150

FrankYang0529 · 2024-09-10T11:48:06Z

When committing all consumed offsets (sync, async, or on close), the new consumer retrieves the offsets from subscriptionState.allConsumed() in the app thread. We should consider to retrieve the allConsumed in the background when processing the events, to avoid inconsistencies given that the subscription state could be modified in the background thread since the moment the allConsumed was retrieved in the app thread.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

lianetm · 2024-09-13T13:47:58Z

Hey @FrankYang0529, thanks for taking this one! I see it is marked as Draft still so haven't looked into it but let me know when ready for review and I'll do. Thanks!

FrankYang0529 · 2024-09-17T02:04:09Z

Hi @lianetm, yes, this one is ready for reviewing. Thank you.

frankvicky

Hi @FrankYang0529
I have a minor comment for consideration, PTAL

frankvicky · 2024-09-18T14:41:36Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+            verify(metadata, never()).updateLastSeenEpochIfNewer(tp, 1);
+            verify(commitRequestManager, never()).commitSync(offsets, deadlineMs);


nit: I noticed that the handling logic when groupId is absent is almost identical across multiple tests. Maybe you could extract a helper method like verifyNoInteractions(metadata, commitRequestManager, isSynced). WDYT ?

lianetm

Hey @FrankYang0529 , thanks for the patch! Took a first pass.

lianetm · 2024-09-26T13:14:07Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

-        CommitRequestManager manager = requestManagers.commitRequestManager.get();
-        CompletableFuture<Void> future = manager.commitAsync(event.offsets());
-        future.whenComplete(complete(event.future()));
+        process((CommitEvent) event);


We have 2 separate event types (async and sync), to then join them together in one here, to then split them again for the actual process with:

if (event.type() == Type.COMMIT_ASYNC) { future = manager.commitAsync(offsets); } else { future = manager.commitSync(offsets, event.deadlineMs()); }

I get that with this we can reuse a bit but wonder if it's worth the twisted flow. Could we maybe keep them separate (as they originally are when the events are created), then process(Sync) that ends up calling the mgr.commitSync, and process(Async) calling manager.commitAsync, and just encapsulate in funcs what we want to reuse in both? (ex. maybeUpdateLastSeenEpochIfNewer() with lines 188-191 that would be called from both, similar for the logic to retrieve offsets from the event, ln 180-181). What do you think?

lianetm · 2024-09-26T13:18:16Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

@@ -208,6 +206,110 @@ public void testSeekUnvalidatedEventWithException() {
        assertInstanceOf(IllegalStateException.class, e.getCause());
    }

+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testSyncCommitEventWithOffsets(boolean withGroupId) {


since this is testing commit it does need a group id, so it should be only for withGroupId=true I expect

lianetm · 2024-09-26T13:18:43Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testSyncCommitEventWithCommitAllConsumed(boolean withGroupId) {


same, only relevant for withGroupId=true right? (and the all the other commit tests down below)

kirktrue

Thanks for the PR @FrankYang0529!

What do you think about wrapping the offset map in an Optional instead of requiring the map to be empty and the flag to be set?

Something like:

    Optional<Map<TopicPartition, OffsetAndMetadata>> = Optional.empty();

IMO, it makes the intent a little more clear. Just a suggestion.

Thanks!

FrankYang0529 · 2024-09-27T14:03:07Z

Hi @lianetm and @kirktrue, thanks for the review. I may need more time for this. I found that AsyncKafkaConsumer#interceptors consume offsets. If we use background thread to get subscriptions.allConsumed(), we need CommitEvent#future to return offsets, so AsyncKafkaConsumer#interceptors doesn't get empty offsets.

FrankYang0529 · 2024-10-07T09:31:16Z

Hi @lianetm and @kirktrue, I addressed all comments. The PR is ready to review now. Thanks.

kirktrue

Thanks @FrankYang0529 for the PR. It looks pretty straightforward, but I had a couple of questions first.

kirktrue · 2024-10-08T23:16:49Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+    private void maybeUpdateLastSeenEpochIfNewer(final Map<TopicPartition, OffsetAndMetadata> offsets) {
+        offsets.forEach((topicPartition, offsetAndMetadata) -> {
+            if (offsetAndMetadata.leaderEpoch().isPresent())
+                metadata.updateLastSeenEpochIfNewer(topicPartition, offsetAndMetadata.leaderEpoch().get());
+        });
    }


The logic in ApplicationEventProcessor.maybeUpdateLastSeenEpochIfNewer() looks very similar to AsyncKafkaConsumer.updateLastSeenEpochIfNewer(). Is it possible to extract that into a shared method somewhere, say ConsumerUtils?

Agree, I extract a new function ConsumerUtils#maybeUpdateLastSeenEpochIfNewer. Thanks for the suggestion.

kirktrue · 2024-10-08T23:19:40Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

-        assertDoesNotThrow(() -> commitEvent.future().complete(null));
+        assertTrue(commitEvent.offsets().isPresent());
+        assertEquals(offsets, commitEvent.offsets().get());
+        assertDoesNotThrow(() -> commitEvent.future().complete(offsets));


I know this line is in the original code, but I have a question: in what case would calling CompletableFuture.complete() possibly throw an exception?

Agree, we should not test Java implementation which is not part of Kafka code. Remove assertDoesNotThrow for CompletableFuture.complete(). Thank you.

kirktrue · 2024-10-08T23:21:00Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

-        verify(metadata).updateLastSeenEpochIfNewer(t0, 2);
-        verify(metadata).updateLastSeenEpochIfNewer(t1, 1);


Out of curiosity, why don't we want to verify the metadata was updated? That still happens in the ApplicationEventProcessor, right?

We mock ApplicationEventProcessor in AsyncKafkaConsumerTest, so can't verify related implementation in it. We test it in https://github.com/apache/kafka/pull/17150/files#diff-87404aabc3e6c99569d678ff904586ab232577240d9932839c52135a34b2b57aR260.

agree with removing the checks on metadata, but then I think we should remove the testCommitSyncLeaderEpochUpdate completely. The metadata update is not done in this component anymore, so this ends up simply testing commitSync, which is already covered. (coverage for the leader epoch update is now in the commitReqMgr, which does the metadata update)

kirktrue · 2024-10-08T23:21:21Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

-        verify(metadata).updateLastSeenEpochIfNewer(t0, 2);
-        verify(metadata).updateLastSeenEpochIfNewer(t1, 1);


Same question here as above.

kirktrue · 2024-10-08T23:25:58Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+
+        setupProcessor(true);
+        doReturn(true).when(metadata).updateLastSeenEpochIfNewer(tp, 1);
+        doReturn(CompletableFuture.completedFuture(null)).when(commitRequestManager).commitSync(offsets, deadlineMs);


Sorry to be daft, but can you explain why we'd want to complete the Future with null instead of result? I'm confused how the call to event.future.get() ends up with the correct value 🤔

The future from CommitRequestManager#commitSync is CompletableFuture<Void>, so it's ok to use null.

The event.future.get() can get expected result is because it gets offsets here https://github.com/apache/kafka/pull/17150/files#diff-583aa43613edc803471d93ed124d1cbed165768ed674c747784ab068e0ed7518R233.

hachikuji · 2024-10-10T20:50:08Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

        acquireAndEnsureOpen();
        long commitStart = time.nanoseconds();
        try {
            SyncCommitEvent syncCommitEvent = new SyncCommitEvent(offsets, calculateDeadlineMs(time, timeout));
-            CompletableFuture<Void> commitFuture = commit(syncCommitEvent);
+            CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> commitFuture = commit(syncCommitEvent);

            Timer requestTimer = time.timer(timeout.toMillis());
            awaitPendingAsyncCommitsAndExecuteCommitCallbacks(requestTimer, true);


Not really clear to me what point this serves. Why do we care about the async commits if they were all sent prior to the sync commit?

Hi @hachikuji, this line was introduced in #15613.

/** * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. */ @Override public void commitSync() { delegate.commitSync(); }

lianetm

Thanks for the updates @FrankYang0529 ! Left some comments, and there is a bit we're missing I believe: on the consumer close, we trigger a commit of all consumed offsets too (still retrieving them directly from the subscription state in the app thread), let's update that one too please.

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

Lines 1302 to 1303 in 984777f

    
           void commitSyncAllConsumed(final Timer timer) { 
        
               Map<TopicPartition, OffsetAndMetadata> allConsumed = subscriptions.allConsumed();

Thanks!

lianetm · 2024-10-29T11:04:06Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

@@ -911,7 +911,7 @@ public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition
            wakeupTrigger.setActiveTask(event.future());
            try {
                final Map<TopicPartition, OffsetAndMetadata> committedOffsets = applicationEventHandler.addAndGet(event);
-                committedOffsets.forEach(this::updateLastSeenEpochIfNewer);
+                ConsumerUtils.maybeUpdateLastSeenEpochIfNewer(metadata, committedOffsets);


The metadata updates have been progressively moved to the background. Should we do the last push here and move this call to maybeUpdateLastSeenEpochIfNewer logic to the background ApplicationEventProcessor? (it's called from there already for all commit events, if we move this and the call in seek to their events process call we would probably be done with it and have it consistently updated only in the background)

lianetm · 2024-10-29T11:09:02Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java

@@ -258,4 +258,11 @@ public static KafkaException maybeWrapAsKafkaException(Throwable t, String messa
        else
            return new KafkaException(message, t);
    }
+
+    public static void maybeUpdateLastSeenEpochIfNewer(ConsumerMetadata metadata, final Map<TopicPartition, OffsetAndMetadata> offsets) {


Once we have all the calls to this metadata update consistently in the background, it will probably make sense to move this helper closer to AppEventProcessor where it's used, what do you think?

lianetm · 2024-10-29T11:12:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

        acquireAndEnsureOpen();
        try {
            AsyncCommitEvent asyncCommitEvent = new AsyncCommitEvent(offsets);
-            lastPendingAsyncCommit = commit(asyncCommitEvent).whenComplete((r, t) -> {
+            lastPendingAsyncCommit = commit(asyncCommitEvent).whenComplete((committedOffsets, t) -> {


nit: should we rename 't' to error or something clearer? just like you did with 'r'

lianetm · 2024-10-29T12:51:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/SyncCommitEvent.java

@@ -20,14 +20,15 @@
 import org.apache.kafka.common.TopicPartition;

 import java.util.Map;
+import java.util.Optional;

 /**
 * Event to commit offsets waiting for a response and retrying on expected retriable errors until
 * the timer expires.


maybe worth adding that if the event contains no offsets it will commit all consumed offsets retrieved from the subscription state, makes sense?

lianetm · 2024-10-29T14:14:29Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+        SyncCommitEvent event = new SyncCommitEvent(Optional.of(offsets), deadlineMs);
+
+        setupProcessor(true);
+        doReturn(true).when(metadata).updateLastSeenEpochIfNewer(tp, 1);


is this line needed? I would expect that we only care about verifying that the metadata was updated, done on ln 260 (no need to return any specific value)

lianetm · 2024-10-29T14:29:57Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+
+        setupProcessor(true);
+        doReturn(offsets).when(subscriptionState).allConsumed();
+        doReturn(true).when(metadata).updateLastSeenEpochIfNewer(tp, 1);


lianetm · 2024-10-29T14:30:06Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+        AsyncCommitEvent event = new AsyncCommitEvent(Optional.of(offsets));
+
+        setupProcessor(true);
+        doReturn(true).when(metadata).updateLastSeenEpochIfNewer(tp, 1);


lianetm · 2024-10-29T14:32:30Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+
+        setupProcessor(true);
+        doReturn(offsets).when(subscriptionState).allConsumed();
+        doReturn(true).when(metadata).updateLastSeenEpochIfNewer(tp, 1);


lianetm · 2024-10-29T14:37:08Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+        setupProcessor(true);
+
+        processor.process(event);
+        Map<TopicPartition, OffsetAndMetadata> committedOffsets = assertDoesNotThrow(() -> event.future().get());


what about verifying that we don't generate requests for empty offsets?

verify(commitRequestManager, never()).commitSync(any(), anyLong());

(let's add it also to the other tests that play with the different commits and empty offsets)

FrankYang0529 · 2024-10-30T14:27:49Z

Hi @lianetm, thanks for the review. I address all comments with following update:

Move ConsumerUtils#maybeUpdateLastSeenEpochIfNewer to ApplicationEventProcessor#maybeUpdateLastSeenEpochIfNewer and do all metadata update in the background thread.
Add testFetchCommittedOffsetsEvent to ApplicationEventProcessorTest.
When autoCommitEnabled is true, using Optional.empty for commitSync function. Update AsyncKafkaConsumerTest#testAutoCommitSyncEnabled to check SyncCommitEvent#offsets is not present.
Update default auto commit enabled to false in AsyncKafkaConsumerTest. Before this PR, we also wait for SyncCommitEvent and we didn't mock complete SyncCommitEvent, but previous test result didn't be blocked on close function by the event. If the default auto commit enabled is true and the test case calls close function evidently, it may be blocked by SyncCommitEvent and wait for 30 seconds. I'm not sure whether the root cause is that we change the future result from Void to Map<TopicPartition, OffsetAndMetadata>.

kirktrue

Thanks for the PR @FrankYang0529!

Overall looks good.

Can we move the bulk of the logic that was added in ApplicationEventProcessor methods into a new method in CommitRequestManager? It seems like we might be able to consolidate the code because the only difference (that I see) between the two is that one calls the CommitRequestManager’s commitAsync() method and the other calls commitSync().

Thanks!

kirktrue · 2024-10-30T20:43:56Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

        } catch (Exception e) {
            // consistent with async auto-commit failures, we do not propagate the exception
-            log.warn("Synchronous auto-commit of offsets {} failed: {}", allConsumed, e.getMessage());
+            log.warn("Synchronous auto-commit failed: {}", e.getMessage());


This is unrelated to this PR, but should we log the stack trace of the error? In commitAsync() we log the stack trace, not just the message.

Suggested change

log.warn("Synchronous auto-commit failed: {}", e.getMessage());

log.warn("Synchronous auto-commit failed", e);

kirktrue · 2024-10-30T20:49:23Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+        Map<TopicPartition, OffsetAndMetadata> offsets = event.offsets().isPresent() ?
+                event.offsets().get() : subscriptions.allConsumed();


Minor, but in my opinion this is a little clearer:

Suggested change

Map<TopicPartition, OffsetAndMetadata> offsets = event.offsets().isPresent() ?

event.offsets().get() : subscriptions.allConsumed();

Map<TopicPartition, OffsetAndMetadata> offsets = event.offsets().orElseGet(subscriptions::allConsumed);

kirktrue · 2024-10-30T20:50:46Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -187,22 +188,62 @@ private void process(final CreateFetchRequestsEvent event) {

    private void process(final AsyncCommitEvent event) {


Can we move the bulk of this logic into a new CommitRequestManager method? We try to keep the process() methods lightweight.

Hi @kirktrue, do you mean that update CommitRequestManager like following?

public CompletableFuture<Void> commitSync(final Optional<Map<TopicPartition, OffsetAndMetadata>> offsets, final long deadlineMs) { if (!offsets.isPresent() || offsets.get().isEmpty()) { return CompletableFuture.completedFuture(null); } CompletableFuture<Void> result = new CompletableFuture<>(); OffsetCommitRequestState requestState = createOffsetCommitRequest(offsets.get(), deadlineMs); commitSyncWithRetries(requestState, result); return result; }

If we do that, we have to run maybeUpdateLastSeenEpochIfNewer(offsets) after future is complete. Is there any side effect for this change?

hey @FrankYang0529 . I cannot see any undesired side-effect, and the change actually makes sense to keep the committed offsets related logic in the commitRequestManager.

I expect we just need to pass the metadata object into the commit manager constructor (it's already available in RequestManagers), and ensure that the commit mgr calls the maybeUpdateLastSeenEpochIfNewer from the fetchCommitted, commitSync and commitAsync (exactly how it's now done in the appEventProcessor)

I was thinking that whether the order of calling maybeUpdateLastSeenEpochIfNewer(offsets) affects the result. After checking again, I think it's okay to move the function to CommitRequestManager. Thanks.

I see, and I would say it's fine if we move it to the commitMgr as you did, as long as we update metadata before completing the result futures that the mgr returns

kirktrue · 2024-10-30T21:02:08Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -508,7 +505,7 @@ public void testSubscriptionRegexEvalOnPollOnlyIfMetadataChanges() {
                mock(ConsumerRebalanceListenerInvoker.class),
                subscriptions,
                "group-id",
-                "client-id");
+                "client-id", false);


Can you move the new false parameter to be aligned with the other parameters?

kirktrue · 2024-10-30T21:02:42Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -846,7 +836,7 @@ public void testUnsubscribeOnClose() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id"));
+            "client-id", false));


Same alignment issue here...

kirktrue · 2024-10-30T21:02:47Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -863,7 +853,7 @@ public void testFailedPartitionRevocationOnClose() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id"));
+            "client-id", false));


Same alignment issue here...

kirktrue · 2024-10-30T21:02:53Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -881,13 +871,17 @@ public void testAutoCommitSyncEnabled() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id");
+            "client-id", false);


Same alignment issue here...

kirktrue · 2024-10-30T21:03:21Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -899,7 +893,7 @@ public void testAutoCommitSyncDisabled() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id");
+            "client-id", false);


Same alignment issue here...

kirktrue · 2024-10-30T21:03:40Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -1700,7 +1690,7 @@ public void testEnsurePollEventSentOnConsumerPoll() {
                mock(ConsumerRebalanceListenerInvoker.class),
                subscriptions,
                "group-id",
-                "client-id");
+                "client-id", false);


Same alignment issue here...

lianetm · 2024-11-01T13:49:58Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+        Map<TopicPartition, OffsetAndMetadata> offsets = event.offsets().orElseGet(subscriptions::allConsumed);
+        if (offsets.isEmpty()) {
+            event.future().complete(Collections.emptyMap());
+            return;
+        }


also this decision of committing all consumed if the commit event has empty offsets could be moved to the commitRequestManager.commitSync to keep all the commit logic there in one place.

lianetm

Hey @FrankYang0529 , thanks for the updates! Regarding this change:

Update default auto commit enabled to false in AsyncKafkaConsumerTest. Before this PR, we also wait for SyncCommitEvent and we didn't mock complete SyncCommitEvent, but previous test result didn't be blocked on close function by the event. If the default auto commit enabled is true and the test case calls close function evidently, it may be blocked by SyncCommitEvent and wait for 30 seconds. I'm not sure whether the root cause is that we change the future result from Void to Map<TopicPartition, OffsetAndMetadata>

I expect that the root cause is that before this PR we completed the commit sync event in the app thread if there were no consumed offsets (so the tests didn't require a mocked completion to close). With this PR we always block on the event, even if no consumed (waiting for the background to check the subscription state). So the failures you faced are expected and I agree we need to update the test. Turning the auto-commit to false as you did makes sense to me given that we have explicit tests for the autoCommit=true combinations.

The only thing we need to review/fix is that there are several tests that call completeCommitSyncApplicationEventSuccessfully but do not perform commitSync or set the auto-commit to true. Those tests don't need to call that anymore I would expect. Could you check and remove it? Thanks!

lianetm · 2024-11-01T14:24:47Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -863,7 +861,8 @@ public void testAutoCommitSyncDisabled() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id");
+            "client-id",
+            false);


not introduced by this PR, but this change made me notice that this test is not really testing what it should. We're verifying that no commit sync event is generated if autoCommit false, but we're only doing consumer.subscribe & seek (which will never generate a sync event anyways). I expect this test would also pass if we pass autoCommit=true.

This test should explicitly call consumer.close after the seek, and then verify that no SyncEvent is generated if autoCommit=false, makes sense?

lianetm · 2024-11-01T14:42:30Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java

@@ -844,14 +837,19 @@ public void testAutoCommitSyncEnabled() {
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
-            "client-id");
+            "client-id",
+            false);


this false is confusing given that the test is called "testAutoCommitSyncEnabled", but I notice it's because the test does not take the config into account and calls the commitSyncAllConsumed explicitly, so what about renaming it to testCommitSyncAllConsumed?

FrankYang0529 · 2024-11-03T14:52:44Z

Hi @lianetm / @kirktrue, thanks for the review and suggestion. I do following change:

Move offsets option checking and maybeUpdateLastSeenEpochIfNewer to CommitRequestManager.
Move related test cases from ApplicationEventProcessorTest to CommitRequestManagerTest.
Remove unnecessary completeCommitSyncApplicationEventSuccessfully from AsyncKafkaConsumerTest.
Update testAutoCommitSyncDisabled to call close explicitly.
Change testAutoCommitSyncEnabled to testCommitSyncAllConsumed.

lianetm

Thanks for the updates @FrankYang0529 ! Here are some more comments.

lianetm · 2024-11-04T20:17:44Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -198,23 +198,47 @@ private void process(final CreateFetchRequestsEvent event) {
    }

    private void process(final AsyncCommitEvent event) {
-        if (!requestManagers.commitRequestManager.isPresent()) {
+        if (requestManagers.commitRequestManager.isEmpty()) {
+            event.future().complete(Map.of());


Should we better complete exceptionally? I expect this would only happen during development phase if a bug is introduced, but still, it would be easier to catch if we make the app event fail because the expected manager is not present. I notice there's no consistency in how we handle these (some process calls complete exceptionally, others just return/complete, but I would lean towards not swallowing the unlikely/unexpected error here)

Thanks for the suggestion. There are some process don't align with the rule. Do we want to create a Jira to handle it? Or I can update them in this PR?

uhm I would leave it out of this PR. Even though there are some other process calls that should probably fail if the expected manager is not present, in some we shouldn't fail and just return/no-op as they do now (ie. commitOnClose no-op if no commit mgr, triggered from close regardless of groupId present or not).

lianetm · 2024-11-04T20:18:27Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

    }

    private void process(final SyncCommitEvent event) {
-        if (!requestManagers.commitRequestManager.isPresent()) {
+        if (requestManagers.commitRequestManager.isEmpty()) {
+            event.future().complete(Map.of());


completeExceptionally?

lianetm · 2024-11-04T20:26:22Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+            future.whenComplete((offsets, throwable) -> {
+                if (throwable != null) {
+                    log.error("Committing offsets failed", throwable);
+                    event.future().completeExceptionally(throwable);
+                } else {
+                    event.future().complete(offsets);
+                }
+            });


could we simplify this to future.whenComplete(complete(event.future())) ?

lianetm · 2024-11-04T20:27:48Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+            future.whenComplete((offsets, throwable) -> {
+                if (throwable != null) {
+                    log.error("Committing offsets failed", throwable);
+                    event.future().completeExceptionally(throwable);
+                } else {
+                    event.future().complete(offsets);
+                }
+            });


could we simplify this to future.whenComplete(complete(event.future()))?

lianetm · 2024-11-04T20:29:21Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

+        future.whenComplete((value, exception) -> {
+            if (exception != null)
+                event.future().completeExceptionally(exception);
+            else {
+                event.future().complete(value);
+            }
+        });


why not reusing future.whenComplete(complete(event.future())) like before?

lianetm · 2024-11-04T21:11:46Z

clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java

+        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(tp, offsetAndMetadata);
+        doReturn(offsets).when(subscriptionState).allConsumed();
+
+        CommitRequestManager commitRequestManager = create(true, 100);


lianetm · 2024-11-04T21:41:33Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

@@ -532,6 +544,7 @@ private void fetchOffsetsWithRetries(final OffsetFetchRequestState fetchRequest,
            }
            if (error == null) {
                result.complete(res);
+                maybeUpdateLastSeenEpochIfNewer(res);


lianetm · 2024-11-04T21:42:07Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

        commitRequest.future.whenComplete((committedOffsets, error) -> {
            if (error != null) {
                asyncCommitResult.completeExceptionally(commitAsyncExceptionForError(error));
            } else {
-                asyncCommitResult.complete(null);
+                asyncCommitResult.complete(commitOffsets);
+                maybeUpdateLastSeenEpochIfNewer(commitOffsets);


this should probably go before completing the "asyncCommitResult" future, to ensure that we don't signal the caller that the commit completed if we haven't updated the metadata yet.

actually, thinking more about this, we shouldn't wait for receiving a response to the request to update the metadata object with the offsets we sent in the request (before this PR and on the classic consumer, we update metadata before sending any request). So this call should really go as soon as we determine the set of commitOffsets (after the early return if (commitOffsets.isEmpty()) maybe?)

lianetm · 2024-11-04T21:42:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

        pendingRequests.addOffsetCommitRequest(requestAttempt);

        // Retry the same commit request while it fails with RetriableException and the retry
        // timeout hasn't expired.
        requestAttempt.future.whenComplete((res, error) -> {
            if (error == null) {
-                result.complete(null);
+                result.complete(requestAttempt.offsets);
+                maybeUpdateLastSeenEpochIfNewer(requestAttempt.offsets);


related to comment from above, should we move this to commitSync, before we send the request or receive a response

lianetm · 2024-11-04T21:44:13Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -187,22 +188,62 @@ private void process(final CreateFetchRequestsEvent event) {

    private void process(final AsyncCommitEvent event) {


I see, and I would say it's fine if we move it to the commitMgr as you did, as long as we update metadata before completing the result futures that the mgr returns

lianetm

Thanks for the updates @FrankYang0529 ! A few more comments.

lianetm · 2024-11-05T17:18:06Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

        commitRequest.future.whenComplete((committedOffsets, error) -> {
            if (error != null) {
                asyncCommitResult.completeExceptionally(commitAsyncExceptionForError(error));
            } else {
-                asyncCommitResult.complete(null);
+                asyncCommitResult.complete(commitOffsets);
+                maybeUpdateLastSeenEpochIfNewer(commitOffsets);


actually, thinking more about this, we shouldn't wait for receiving a response to the request to update the metadata object with the offsets we sent in the request (before this PR and on the classic consumer, we update metadata before sending any request). So this call should really go as soon as we determine the set of commitOffsets (after the early return if (commitOffsets.isEmpty()) maybe?)

lianetm · 2024-11-05T17:19:45Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

        pendingRequests.addOffsetCommitRequest(requestAttempt);

        // Retry the same commit request while it fails with RetriableException and the retry
        // timeout hasn't expired.
        requestAttempt.future.whenComplete((res, error) -> {
            if (error == null) {
-                result.complete(null);
+                result.complete(requestAttempt.offsets);
+                maybeUpdateLastSeenEpochIfNewer(requestAttempt.offsets);


related to comment from above, should we move this to commitSync, before we send the request or receive a response

lianetm · 2024-11-05T18:35:25Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

@@ -72,6 +72,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;

+@SuppressWarnings("ClassDataAbstractionCoupling")


could you double check if we really need this after the latest changes? (I don't see many new deps in the changes anymore)

Yes, we still need this line, or it shows following error.

[ant:checkstyle] [ERROR] apache/kafka/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:75:1: Class Data Abstraction Coupling is 27 (max allowed is 25) classes [ApplicationEventProcessor, AssignmentChangeEvent, AsyncCommitEvent, CheckAndUpdatePositionsEvent, CommitOnCloseEvent, CompletableFuture, CreateFetchRequestsEvent, FetchCommittedOffsetsEvent, ListOffsetsEvent, LogContext, MockRebalanceListener, MockTime, NetworkClientDelegate.PollResult, OffsetAndMetadata, PollEvent, RequestManagers, ResetOffsetEvent, SeekUnvalidatedEvent, SubscriptionState, SubscriptionState.FetchPosition, SyncCommitEvent, TopicMetadataEvent, TopicPartition, TopicPatternSubscriptionChangeEvent, TopicSubscriptionChangeEvent, UnsubscribeEvent, UpdatePatternSubscriptionEvent]. [ClassDataAbstractionCoupling]

lianetm · 2024-11-06T15:49:20Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

-        if (!requestManagers.commitRequestManager.isPresent()) {
+        if (requestManagers.commitRequestManager.isEmpty()) {
+            event.future().completeExceptionally(new KafkaException("Unable to async commit " +
+                "offset because the CommittedRequestManager is not available. Check if group.id was set correctly"));


typo: CommitRequestManager

lianetm · 2024-11-06T15:49:26Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

-        if (!requestManagers.commitRequestManager.isPresent()) {
+        if (requestManagers.commitRequestManager.isEmpty()) {
+            event.future().completeExceptionally(new KafkaException("Unable to sync commit " +
+                "offset because the CommittedRequestManager is not available. Check if group.id was set correctly"));


typo: CommitRequestManager

lianetm · 2024-11-06T15:50:02Z

.../main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

    }

    private void process(final FetchCommittedOffsetsEvent event) {
-        if (!requestManagers.commitRequestManager.isPresent()) {
+        if (requestManagers.commitRequestManager.isEmpty()) {
            event.future().completeExceptionally(new KafkaException("Unable to fetch committed " +
                    "offset because the CommittedRequestManager is not available. Check if group.id was set correctly"));


CommitRequestManager

lianetm · 2024-11-06T15:52:24Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+        ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
+        assertInstanceOf(IllegalStateException.class, e.getCause());


assertFutureThrows?

lianetm · 2024-11-06T15:52:44Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+        ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
+        assertInstanceOf(IllegalStateException.class, e.getCause());


assertFutureThrows?

lianetm

Just minor suggestions left. Also please merge trunk latest changes if you haven't. Thanks!

FrankYang0529 · 2024-11-06T16:08:08Z

Hi @lianetm, thanks for the review. I addressed all comments and rebased trunk.

lianetm · 2024-11-06T16:13:54Z

...t/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java

+
+        setupProcessor(false);
+        processor.process(event);
+        assertThrows(ExecutionException.class, () -> event.future().get());


this should also assertFutureThrows(KafkaException) right? (and nit/taste, I would import TestUtils.assertFutureThrows and simplify the code to have calls to assertFutureThrows)

… in background thread Signed-off-by: PoAn Yang <[email protected]>

lianetm

Thanks for all the updates @FrankYang0529 ! LGTM.

FrankYang0529 · 2024-11-07T14:04:26Z

Thanks for your review and patience. 🙏

lianetm added consumer ctr Consumer Threading Refactor (KIP-848) labels Sep 11, 2024

FrankYang0529 force-pushed the KAFKA-17480 branch from 68a6522 to 4951c74 Compare September 11, 2024 12:39

FrankYang0529 force-pushed the KAFKA-17480 branch 2 times, most recently from 4576cc6 to 6622353 Compare September 17, 2024 01:56

FrankYang0529 marked this pull request as ready for review September 17, 2024 02:03

frankvicky reviewed Sep 18, 2024

View reviewed changes

lianetm reviewed Sep 26, 2024

View reviewed changes

kirktrue added KIP-848 The Next Generation of the Consumer Rebalance Protocol clients labels Sep 26, 2024

kirktrue reviewed Sep 26, 2024

View reviewed changes

FrankYang0529 marked this pull request as draft September 27, 2024 14:03

FrankYang0529 force-pushed the KAFKA-17480 branch 3 times, most recently from 451b3c6 to e43c2e4 Compare October 7, 2024 07:10

FrankYang0529 marked this pull request as ready for review October 7, 2024 07:10

FrankYang0529 requested review from kirktrue and lianetm October 7, 2024 09:31

kirktrue reviewed Oct 8, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from e43c2e4 to c71a86e Compare October 10, 2024 15:11

hachikuji reviewed Oct 10, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from c71a86e to b19433b Compare October 21, 2024 00:05

lianetm reviewed Oct 29, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from b19433b to a6f0d33 Compare October 30, 2024 13:21

kirktrue suggested changes Oct 30, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch 2 times, most recently from 9edef7b to a1ac1c0 Compare November 1, 2024 07:07

lianetm reviewed Nov 1, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from a1ac1c0 to 98e5286 Compare November 3, 2024 14:47

lianetm reviewed Nov 4, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch 2 times, most recently from b31521b to feec864 Compare November 5, 2024 14:41

lianetm reviewed Nov 5, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from feec864 to a9c2ee3 Compare November 6, 2024 11:15

lianetm reviewed Nov 6, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17480 branch from a9c2ee3 to 2e836dd Compare November 6, 2024 16:07

lianetm reviewed Nov 6, 2024

View reviewed changes

KAFKA-17480: New consumer commit all consumed should retrieve offsets…

7d2def0

… in background thread Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 force-pushed the KAFKA-17480 branch from 2e836dd to 7d2def0 Compare November 7, 2024 12:09

lianetm approved these changes Nov 7, 2024

View reviewed changes

lianetm merged commit b213c64 into apache:trunk Nov 7, 2024
8 checks passed

FrankYang0529 deleted the KAFKA-17480 branch November 7, 2024 14:46

		verify(metadata, never()).updateLastSeenEpochIfNewer(tp, 1);
		verify(commitRequestManager, never()).commitSync(offsets, deadlineMs);

		verify(metadata).updateLastSeenEpochIfNewer(t0, 2);
		verify(metadata).updateLastSeenEpochIfNewer(t1, 1);

	void commitSyncAllConsumed(final Timer timer) {
	Map<TopicPartition, OffsetAndMetadata> allConsumed = subscriptions.allConsumed();

	log.warn("Synchronous auto-commit failed: {}", e.getMessage());
	log.warn("Synchronous auto-commit failed", e);

		Map<TopicPartition, OffsetAndMetadata> offsets = event.offsets().isPresent() ?
		event.offsets().get() : subscriptions.allConsumed();

		@@ -187,22 +188,62 @@ private void process(final CreateFetchRequestsEvent event) {

		private void process(final AsyncCommitEvent event) {

		ExecutionException e = assertThrows(ExecutionException.class, () -> event.future().get());
		assertInstanceOf(IllegalStateException.class, e.getCause());

KAFKA-17480: New consumer commit all consumed should retrieve offsets in background thread #17150

KAFKA-17480: New consumer commit all consumed should retrieve offsets in background thread #17150

Conversation

FrankYang0529 commented Sep 10, 2024 • edited Loading

Committer Checklist (excluded from commit message)

lianetm commented Sep 13, 2024

FrankYang0529 commented Sep 17, 2024

frankvicky left a comment

Choose a reason for hiding this comment

frankvicky Sep 18, 2024 • edited Loading

Choose a reason for hiding this comment

lianetm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirktrue left a comment

Choose a reason for hiding this comment

FrankYang0529 commented Sep 27, 2024

FrankYang0529 commented Oct 7, 2024

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm Oct 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented Oct 30, 2024 • edited Loading

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented Nov 3, 2024

lianetm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented Sep 10, 2024 •

edited

Loading

frankvicky Sep 18, 2024 •

edited

Loading

lianetm left a comment •

edited

Loading

lianetm Oct 29, 2024 •

edited

Loading

FrankYang0529 commented Oct 30, 2024 •

edited

Loading

lianetm left a comment •

edited

Loading