KAFKA-17505: New consumer seekToBeginning/End should run in background thread #17230

TaiJuWu · 2024-09-19T01:24:35Z

Jira: https://issues.apache.org/jira/browse/KAFKA-17505

Move seekToBeginning/End run in background thread.
The app thread will be blocked until background finish this operation in order to avoid race issue.

Committer Checklist (excluded from commit message)

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

chia7712 · 2024-09-20T13:50:47Z

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

-            Collection<TopicPartition> parts = partitions.isEmpty() ? subscriptions.assignedPartitions() : partitions;
-            subscriptions.requestOffsetReset(parts, OffsetResetStrategy.LATEST);
+            cachedSubscriptionHasAllFetchPositions = false;
+            applicationEventHandler.add(new ResetOffsetEvent(partitions, OffsetResetStrategy.LATEST));


seekToBeginning and seekToEnd don't wait the completion, but seek does. not sure why those similar methods have different behavior. The doc of seek declares the effect will be applied on next poll, and ClassicKafkaConsumer#seek does not wait also. IMHO, AsyncKafkaConsumer#seek should use add rather than addAndGet.

https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L792

@kirktrue @lianetm WDYT?

open https://issues.apache.org/jira/browse/KAFKA-17586

Hey here, sorry for the late reply. I agree that these 3 methods should have similar behaviour, but I wonder if the case is that they should all use addAndGet? My concern is, when seeking (either of the 3 methods), we do expect that as soon as they complete, the position is updated so that other funcs will see it (ex. currentLag, poll). We would be not respecting that anymore if we use add. seek could complete without having updated the offsets right?

From the seek contract, it states that "Overrides the fetch offsets that the consumer will use on the next {@link #poll(Duration) poll(timeout)}", so we need to:

override fetch offsets (this requires addAndGet I expect)

make sure they are available on the next poll (again addAndGet on seek, otherwise consumer.seek + consumer.poll may not find the offsets if the poll happens when the seek with add is still being processed in the background right?). Also calls like currentLag after that seek + add could not have the info they need I guess.

Thoughts?

@lianetm thanks for the sharing. you are right, those seek methods need to wait the event gets completed to avoid race condition. +1 to use addAndGet on seek, seekToBeginning, and seekToEnd. I will close https://issues.apache.org/jira/browse/KAFKA-17586

@TaiJuWu Could you please use addAndGet instead?

Another option we've used elsewhere is to stash a reference to the ResetOffsetsEvent in the AsyncKafkaConsumer after it's enqueued so it can be checked later (i.e. in poll()).

I will admit I don't like the idea 🤔

Hi @chia7712 @lianetm , thanks for your review and sharing.
I will update this PR according your suggestion.
Thanks a lot.

I noticed another difference between seek(), seekToBeginning(), and seekToEnd()—seek() sets the position and validates it (if possible) while the other two forms request the positions be reset. So even if the code in seekToBeginning() and seekToEnd() is switched to addAndGet(), it doesn't guarantee that the positions have been reset when the method exits, does it?

I noticed another difference between seek(), seekToBeginning(), and seekToEnd()—seek() sets the position and validates it (if possible) while the other two forms request the positions be reset. So even if the code in seekToBeginning() and seekToEnd() is switched to addAndGet(), it doesn't guarantee that the positions have been reset when the method exits, does it?

Hi @kirktrue , you are right and thanks for your detailed review.
I think seekToBeginning and seekToEnd can be guarantee by follow steps if we leverage offsetManager:
updateFetchPositions[1] -> updatePositionsWithOffsets[2] -> initWithPartitionOffsetsIfNeeded[3] -> resetPositionsIfNeeded[4] -> sendListOffsetsRequestsAndResetPositions[5]

FrankYang0529

Hi @TaiJuWu, thanks for the PR. Leave some minor comments.

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

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

kirktrue

Thanks for the PR @TaiJuWu!

I left a few comments, mostly minor tweaks.

Thanks!

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

kirktrue · 2024-09-26T23:18:39Z

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

-            Collection<TopicPartition> parts = partitions.isEmpty() ? subscriptions.assignedPartitions() : partitions;
-            subscriptions.requestOffsetReset(parts, OffsetResetStrategy.LATEST);
+            cachedSubscriptionHasAllFetchPositions = false;
+            applicationEventHandler.add(new ResetOffsetEvent(partitions, OffsetResetStrategy.LATEST));


Another option we've used elsewhere is to stash a reference to the ResetOffsetsEvent in the AsyncKafkaConsumer after it's enqueued so it can be checked later (i.e. in poll()).

I will admit I don't like the idea 🤔

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

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

kirktrue · 2024-09-26T23:31:59Z

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

-            Collection<TopicPartition> parts = partitions.isEmpty() ? subscriptions.assignedPartitions() : partitions;
-            subscriptions.requestOffsetReset(parts, OffsetResetStrategy.LATEST);
+            cachedSubscriptionHasAllFetchPositions = false;
+            applicationEventHandler.add(new ResetOffsetEvent(partitions, OffsetResetStrategy.LATEST));


I noticed another difference between seek(), seekToBeginning(), and seekToEnd()—seek() sets the position and validates it (if possible) while the other two forms request the positions be reset. So even if the code in seekToBeginning() and seekToEnd() is switched to addAndGet(), it doesn't guarantee that the positions have been reset when the method exits, does it?

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

kirktrue · 2024-09-26T23:34:39Z

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

+    public void testSeekToBeginning() {
+        SubscriptionState subscriptions = mock(SubscriptionState.class);
+        TopicPartition topic = new TopicPartition("test", 0);
+        consumer = spy(newConsumer(


Does consumer need to be wrapped as a spy in this case? It doesn't look like we're verifying or modifying it in any way 🤔

You are right, we do not this, thanks.

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

TaiJuWu · 2024-09-30T00:50:29Z

Hi @kirktrue @lianetm @chia7712,

I have updated the seekToBeginning and seekToEnd methods to use addAndGet. Please take a look.

init version

9f39191

frankvicky added consumer ctr Consumer Threading Refactor (KIP-848) labels Sep 19, 2024

chia7712 reviewed Sep 20, 2024

View reviewed changes

add comments

ffa5fb3

github-actions bot added the clients label Sep 21, 2024

add test

d6a4696

TaiJuWu marked this pull request as ready for review September 21, 2024 05:50

FrankYang0529 reviewed Sep 21, 2024

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java Outdated Show resolved Hide resolved

clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java Outdated Show resolved Hide resolved

fix error comments and test

823951d

TaiJuWu force-pushed the gh-KAFKA-17505 branch from e8130ae to 823951d Compare September 21, 2024 19:01

fix test error and add seekToEnd

1bdfb9f

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

kirktrue suggested changes Sep 26, 2024

View reviewed changes

TaiJuWu added 2 commits September 28, 2024 20:57

use blocking call in seekToXXX

7ee1ed3

address comments

c732950

chia7712 added the ci-approved label Sep 28, 2024

TaiJuWu added 2 commits September 29, 2024 09:23

add more test

8e10de1

Remove offsetManager in ResetOffsetEvent

f8379a0

TaiJuWu force-pushed the gh-KAFKA-17505 branch from 32f7cab to f8379a0 Compare September 29, 2024 04:17

TaiJuWu added 2 commits September 29, 2024 12:24

refactor seekToEnd/Beginning

b8b9e27

revert unecessary change

2772942

TaiJuWu force-pushed the gh-KAFKA-17505 branch from 1820990 to 2772942 Compare September 29, 2024 04:39

TaiJuWu added 3 commits September 29, 2024 16:01

fix testResetOffsetEvent

2ddc429

Merge branch 'trunk' into gh-KAFKA-17505

70cc0d7

improve testResetOffset

30b8982

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17505: New consumer seekToBeginning/End should run in background thread #17230

KAFKA-17505: New consumer seekToBeginning/End should run in background thread #17230

TaiJuWu commented Sep 19, 2024 •

edited

Loading

chia7712 Sep 20, 2024

chia7712 Sep 20, 2024

lianetm Sep 26, 2024 •

edited

Loading

chia7712 Sep 26, 2024

chia7712 Sep 26, 2024

kirktrue Sep 26, 2024

TaiJuWu Sep 26, 2024

kirktrue Sep 26, 2024

TaiJuWu Sep 28, 2024 •

edited

Loading

FrankYang0529 left a comment

kirktrue left a comment

kirktrue Sep 26, 2024

kirktrue Sep 26, 2024

kirktrue Sep 26, 2024

TaiJuWu Sep 28, 2024

TaiJuWu commented Sep 30, 2024

KAFKA-17505: New consumer seekToBeginning/End should run in background thread #17230

Are you sure you want to change the base?

KAFKA-17505: New consumer seekToBeginning/End should run in background thread #17230

Conversation

TaiJuWu commented Sep 19, 2024 • edited Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm Sep 26, 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

TaiJuWu Sep 28, 2024 • edited Loading

Choose a reason for hiding this comment

FrankYang0529 left a comment

Choose a reason for hiding this comment

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

TaiJuWu commented Sep 30, 2024

TaiJuWu commented Sep 19, 2024 •

edited

Loading

lianetm Sep 26, 2024 •

edited

Loading

TaiJuWu Sep 28, 2024 •

edited

Loading