KAFKA-17448: New consumer seek should update positions in background thread #17075

FrankYang0529 · 2024-09-03T13:04:40Z

The AsyncKafkaConsumer#seek uses SubscriptionState#seekUnvalidated which calls SubscriptionState#assignedState. If we call it in app thread, it may have race condition with background thread. Move SubscriptionState#seekUnvalidated to background thread to avoid concurrent write.

Committer Checklist (excluded from commit message)

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

kirktrue

Thanks for the PR, @FrankYang0529!

I think the implementation is sound. I just have the one suggestion to defer the creation of the FetchPosition until the ApplicationEventProcessor.process() invocation. This makes the calls in AsyncKafkaConsumer cleaner (IMO), and also keeps the call to ConsumerMetadata.currentLeader() as close as possible to the seekUnvalidated() call.

Thanks!

kirktrue · 2024-09-03T16:05:16Z

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

+
+public class SeekUnvalidatedEvent extends CompletableApplicationEvent<Void> {
+    private final TopicPartition partition;
+    private final SubscriptionState.FetchPosition position;


It's just my preference, but I'd rather not pass the FetchPosition around in the event. Could we instead include the following directly in the event:

TopicPartition partition

long offset

Optional<Integer> offsetEpoch

The ApplicationEventProcessor has access to the ConsumerMetadata object, so we could fetch the current leader there, in one place, instead of having it twice in the AsyncKafkaConsumer.

kirktrue · 2024-09-03T16:08:38Z

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

+
+    private void process(final SeekUnvalidatedEvent event) {
+        try {
+            subscriptions.seekUnvalidated(event.partition(), event.position());


If we update the event class, we can then move more of the logic here:

Suggested change

subscriptions.seekUnvalidated(event.partition(), event.position());

SubscriptionState.FetchPosition newPosition = new SubscriptionState.FetchPosition(

event.offset(),

event.offsetEpoch(),

metadata.currentLeader(event.partition())

);

subscriptions.seekUnvalidated(event.partition(), newPosition);

kirktrue · 2024-09-03T16:11:22Z

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

+            SeekUnvalidatedEvent seekUnvalidatedEventEvent = new SeekUnvalidatedEvent(calculateDeadlineMs(timer), partition, newPosition);
+            applicationEventHandler.addAndGet(seekUnvalidatedEventEvent);


Per the other suggestions, we could make this a bit more succinct:

Suggested change

SeekUnvalidatedEvent seekUnvalidatedEventEvent = new SeekUnvalidatedEvent(calculateDeadlineMs(timer), partition, newPosition);

applicationEventHandler.addAndGet(seekUnvalidatedEventEvent);

SeekUnvalidatedEvent event = new SeekUnvalidatedEvent(

calculateDeadlineMs(timer),

partition,

offsetAndMetadata.offset(),

offsetAndMetadata.leaderEpoch()

);

applicationEventHandler.addAndGet(event);

kirktrue · 2024-09-03T16:12:28Z

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

+            SeekUnvalidatedEvent seekUnvalidatedEventEvent = new SeekUnvalidatedEvent(calculateDeadlineMs(timer), partition, newPosition);
+            applicationEventHandler.addAndGet(seekUnvalidatedEventEvent);


Per the other suggestions, we could defer the creation of the SubscriptionState.FetchPosition until the ApplicationEventProcessor processor method to make this call a bit more succinct:

Suggested change

SeekUnvalidatedEvent seekUnvalidatedEventEvent = new SeekUnvalidatedEvent(calculateDeadlineMs(timer), partition, newPosition);

applicationEventHandler.addAndGet(seekUnvalidatedEventEvent);

SeekUnvalidatedEvent event = new SeekUnvalidatedEvent(

calculateDeadlineMs(timer),

partition,

offset,

Optional.empty(), // This will ensure we skip validation

);

applicationEventHandler.addAndGet(event);

kirktrue · 2024-09-03T16:17:39Z

Also, could you put a sentence or two description in the event class’ Javadoc? Thanks!

FrankYang0529 · 2024-09-06T12:26:30Z

@kirktrue, thanks for reviewing and suggestions. I addressed all comments and add Javadoc for the SeekUnvalidatedEvent. Could you take a look when you have time? Thank you.

lianetm · 2024-09-06T14:45:16Z

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

+
+    @Override
+    protected String toStringBase() {
+        return super.toStringBase() + ", partition=" + partition + ", offset=" + offset + ", offsetEpoch=" + offsetEpoch;


given that offsetEpoch is an Optional object, won't this print something like "offsetEpoch=Optional(10)" or "offsetEpoch=Optional.empty"? If so, I find it confusing.

We could check offsetEpoch.isPresent, and only add "offsetsEpoch=.." to the output string if we have a value for it, seems clearer as output. What do you think?

Thanks for the review. Updated it.

lianetm · 2024-09-06T14:51:35Z

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

@@ -183,6 +186,38 @@ public void testAssignmentChangeEventWithException() {
        assertInstanceOf(IllegalStateException.class, e.getCause());
    }

+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})


do we get any value from these params? I expect that having or not a group ID does not change the course of this test that is only about seeking to a position

Yes, I think we don't need to test both situations for this case. Thanks.

lianetm · 2024-09-06T14:51:42Z

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

+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})


lianetm · 2024-09-06T14:52:29Z

Hey @FrankYang0529 , thanks for the patch! Took a full pass, just some minor comments.

lianetm · 2024-09-10T14:58:48Z

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

+        return offsetEpoch
+                .map(integer -> super.toStringBase() + ", partition=" + partition + ", offset=" + offset + ", offsetEpoch=" + integer)
+                .orElseGet(() -> super.toStringBase() + ", partition=" + partition + ", offset=" + offset);


Suggested change

return offsetEpoch

.map(integer -> super.toStringBase() + ", partition=" + partition + ", offset=" + offset + ", offsetEpoch=" + integer)

.orElseGet(() -> super.toStringBase() + ", partition=" + partition + ", offset=" + offset);

return super.toStringBase()

+ ", partition=" + partition

+ ", offset=" + offset

+ offsetEpoch.map(integer -> ", offsetEpoch=" + integer).orElse("");

What about this, just to avoid duplicating what's common?

Agree, I think it's better solution. Thanks for the suggestion. Updated it.

lianetm · 2024-09-10T15:53:11Z

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

+import java.util.Optional;
+
+/**
+ * Event to perform {@link org.apache.kafka.clients.consumer.internals.SubscriptionState#seekUnvalidated(TopicPartition, SubscriptionState.FetchPosition)}


one more detail, we already import SubscriptionState on ln 19, so we can simplify here to ...link SubscriptionState#....

Updated it. Thank you.

…thread Signed-off-by: PoAn Yang <payang@apache.org>

lianetm

LGTM. Thanks!

kirktrue added consumer KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels Sep 3, 2024

kirktrue suggested changes Sep 3, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17448 branch from 9b3c61c to a986d50 Compare September 6, 2024 12:25

lianetm reviewed Sep 6, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17448 branch from a986d50 to 7f56043 Compare September 10, 2024 11:51

lianetm reviewed Sep 10, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17448 branch from 7f56043 to 5b0727a Compare September 10, 2024 15:23

lianetm reviewed Sep 10, 2024

View reviewed changes

KAFKA-17448: New consumer seek should update positions in background …

2ef030b

…thread Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 force-pushed the KAFKA-17448 branch from 5b0727a to 2ef030b Compare September 11, 2024 09:59

lianetm approved these changes Sep 11, 2024

View reviewed changes

lianetm merged commit 3f4c25f into apache:trunk Sep 11, 2024
8 of 9 checks passed

FrankYang0529 deleted the KAFKA-17448 branch September 12, 2024 02:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-17448: New consumer seek should update positions in background thread #17075

KAFKA-17448: New consumer seek should update positions in background thread #17075

FrankYang0529 commented Sep 3, 2024

kirktrue left a comment

kirktrue Sep 3, 2024

kirktrue Sep 3, 2024

kirktrue Sep 3, 2024

kirktrue Sep 3, 2024

kirktrue commented Sep 3, 2024 •

edited

Loading

FrankYang0529 commented Sep 6, 2024

lianetm Sep 6, 2024

FrankYang0529 Sep 10, 2024

lianetm Sep 6, 2024

FrankYang0529 Sep 10, 2024

lianetm Sep 6, 2024

lianetm commented Sep 6, 2024

lianetm Sep 10, 2024

FrankYang0529 Sep 10, 2024

lianetm Sep 10, 2024

FrankYang0529 Sep 11, 2024

lianetm left a comment

-            subscriptions.seekUnvalidated(event.partition(), event.position());
+            SubscriptionState.FetchPosition newPosition = new SubscriptionState.FetchPosition(
+                event.offset(),
+                event.offsetEpoch(),
+                metadata.currentLeader(event.partition())
+            );
+            subscriptions.seekUnvalidated(event.partition(), newPosition);

		SeekUnvalidatedEvent seekUnvalidatedEventEvent = new SeekUnvalidatedEvent(calculateDeadlineMs(timer), partition, newPosition);
		applicationEventHandler.addAndGet(seekUnvalidatedEventEvent);

KAFKA-17448: New consumer seek should update positions in background thread #17075

KAFKA-17448: New consumer seek should update positions in background thread #17075

Conversation

FrankYang0529 commented Sep 3, 2024

Committer Checklist (excluded from commit message)

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

kirktrue commented Sep 3, 2024 • edited Loading

FrankYang0529 commented Sep 6, 2024

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 commented Sep 6, 2024

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

Choose a reason for hiding this comment

kirktrue commented Sep 3, 2024 •

edited

Loading