KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll #17165

FrankYang0529 · 2024-09-11T14:46:21Z

To fulfill "rebalances will only occur during an active call to KafkaConsumer#poll(Duration)", we should not send JoinRequest after AsyncKafkaConsumer#subscribe. Add a flag subscriptionUpdated to AbstractMembershipManager#onSubscriptionUpdated. When calling AsyncKafkaConsumer#subscribe, set the flag to true. When calling AsyncKafkaConsumer#poll, send the JoinRequest if the flag is true.

Committer Checklist (excluded from commit message)

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

TaiJuWu · 2024-09-12T10:31:28Z

Hi @FrankYang0529 , I opened a PR for this behavior.
Could you take a look? Thanks a lot!

#17175

FrankYang0529 · 2024-09-12T12:45:49Z

I opened a PR for this behavior. Could you take a look? Thanks a lot!

#17175

Hi @TaiJuWu, thanks for the PR. I also have a similar test case testPollSendRequestForRebalance. We can also have your test after this PR is merged.

TaiJuWu · 2024-09-13T00:56:26Z

I opened a PR for this behavior. Could you take a look? Thanks a lot!
#17175

Hi @TaiJuWu, thanks for the PR. I also have a similar test case testPollSendRequestForRebalance. We can also have your test after this PR is merged.

Hi @FrankYang0529 , I will close my PR because they do same thing and I leave a comment for testPollSendRequestForRebalance.

TaiJuWu · 2024-09-13T00:53:20Z

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

+        client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, node), node);
+        KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata,
+                assignor, true, groupInstanceId);
+        consumer.subscribe(singletonList(topic));


Maybe we need to make application threadsleep here to make sure network thread does not send join request.
WDYT?

Hi @TaiJuWu, application and network are different threads. I'm not sure why making application thread sleep can let network thread doesn't send join request. Also , this test would like to make sure KafkaConsumer#subscribe doesn't send join request, so we should not have other operations to make the result. Could you elaborate more on your idea? Thank you.

Sorry for the confusion.

The reason for making the application sleep here is to give the network enough time to send the join request even without poll. This helps to reproduce the bug more easily and consistently.

Additionally, to ensure the behavior we’ve promised in the documentation, the two operations (subscribe and poll) should not be too close together in the timeline in this test. Does that make sense?

Hi @TaiJuWu, thanks for the suggestion. It's hard to say waiting how much time can make sure KafkaConsumer#subscribe doesn't send a join request. I add two unit tests to check the behavior.

In ApplicationEventProcessorTest#testSubscriptionChangeEvent, we check ApplicationEventProcessor#process(SubscriptionChangeEvent) doesn't call ConsumerMembershipManager#maybeJoinGroup.
In ConsumerMembershipManagerTest#testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup, we check ConsumerMembershipManager#onSubscriptionUpdated keeps member state as UNSUBSCRIBED and calling transitionToJoining in maybeJoinGroup.

Thanks for your explanation.

lianetm · 2024-09-16T16:30:37Z

Hey @FrankYang0529, so seems this draft is the PR we're keeping right? Let me know when it's ready for review and I'll be happy to take a look. Thanks!

FrankYang0529 · 2024-09-17T02:37:53Z

Hi @lianetm, yes, the PR is ready for review. Thank you.

lianetm

Hey @FrankYang0529 , thanks for the patch! Left some comments for consideration.

lianetm · 2024-09-17T19:48:40Z

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

        membershipManager.onSubscriptionUpdated();
+        assertFalse(membershipManager.subscriptionUpdated());


these 2 lines are a bit confusing (we trigger "onSubscriptionUpdated" and expect "subscriptionUpdated" to be false). I would say that the boolean var name is what we should review, because it truly indicates if the member should join (related to comment on the membershipManager.subscriptionUpdated var)

Change variable from subscriptionUpdated to shouldTransitionToJoining and add more comments for it. Thank you.

lianetm · 2024-09-17T19:56:55Z

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

@@ -1617,12 +1617,25 @@ public void testRevokePartitionsUsesTopicNamesLocalCacheWhenMetadataNotAvailable
    }

    @Test
-    public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+    public void testOnSubscriptionUpdatedDoNothingIfInGroup() {


what about testSubscriptionUpdateDoesNotTransitionToJoining (or similar)? (not transitioning to joining is really the core bit we were missing and are fixing/test here..and is actually consistent with the name you have for the similar test right below)

Thanks for the suggestion. I think whether the member is in the group is important as well, because it determines whether to change shouldTransitionToJoining variable. I update the test case name to testOnSubscriptionUpdatedDoesNotSetShouldTransitionToJoiningIfInGroup and another test case below to testOnSubscriptionUpdatedSetShouldTransitionToJoiningIfNotInGroup. Do you think it makes sense? Thank you.

lianetm · 2024-09-17T20:06:33Z

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

+    public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+        ConsumerMembershipManager membershipManager = createMembershipManager(null);
+        assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+        membershipManager.onSubscriptionUpdated();


should we verify here that there was no transitionToJoining? (to be sure that the transition is only triggered after the call to maybeJoinGroup)

Yes, added it. Thanks.

lianetm · 2024-09-17T20:10:16Z

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -3443,6 +3443,34 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
        }
    }

+    @ParameterizedTest
+    @EnumSource(value = GroupProtocol.class)
+    public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {


typo testPollSendsRequestForRebalance? (and maybe clearer testPollSendsRequestToJoin?)

lianetm · 2024-09-17T20:13:56Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

@@ -189,6 +191,9 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl

    private final Time time;

+    // AtomicBoolean to track if the subscription has been updated


this is truly more to know if we should join after the subscription has been updated right? (true if not in group)

FrankYang0529 · 2024-09-19T12:23:50Z

Hi @lianetm, I found that PlaintextConsumerSubscriptionTest#testSubscribeInvalidTopic is flaky in this PR, so I will convert to draft first. Thanks.

lianetm · 2024-09-19T12:53:08Z

Hey @FrankYang0529, just in case it helps, we have this https://issues.apache.org/jira/browse/KAFKA-17286 related to that flaky test with some ideas in the comments.

FrankYang0529 · 2024-09-19T12:59:42Z

Hey @FrankYang0529, just in case it helps, we have this https://issues.apache.org/jira/browse/KAFKA-17286 related to that flaky test with some ideas in the comments.

Yes, that helps. I found that both in trunk and this PR, the failed result is reproducible by adding Thread.sleep(1000) at the end of PlaintextConsumerSubscriptionTest#testSubscribeInvalidTopic.

Do you think we let https://issues.apache.org/jira/browse/KAFKA-17286 handle this flaky? Or we should solve it in this PR? Thanks.

lianetm · 2024-09-23T11:44:05Z

Hey @FrankYang0529 , sorry for the late reply, I was traveling last week but I'm back, so will be taking another look at this. You can mark it as ready, we can leave the fix for the flaky test separate given that we already understand the root cause not related to this change. I'll take another look at this one between today and tomorrow. Thanks!

lianetm

Hey @FrankYang0529, I took another full pass and left some comments for consideration. Thanks!

lianetm · 2024-09-23T19:47:29Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

@@ -466,6 +474,17 @@ String memberIdInfoForLog() {
     */
    public void onSubscriptionUpdated() {
        if (state == MemberState.UNSUBSCRIBED) {


I wonder if this check is still needed here? It made sense before because we were doing the actual join, but now we're just setting a flag (that looks more like a "subscriptionUpdated" var now). Then on the maybeJoinGroup is where we need to check that the state if UNSUBSCRIBED and subscriptionUpdated then join. Makes sense?

lianetm · 2024-09-23T20:06:55Z

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

@@ -246,7 +252,9 @@ private void process(final SubscriptionChangeEvent ignored) {
     */
    private void process(final UnsubscribeEvent event) {
        if (requestManagers.consumerHeartbeatRequestManager.isPresent()) {
+            System.out.println("UnsubscribeEvent: " + event);


lets remove this please

lianetm · 2024-09-23T20:07:00Z

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

            CompletableFuture<Void> future = requestManagers.consumerHeartbeatRequestManager.get().membershipManager().leaveGroup();
+            System.out.println("UnsubscribeEvent: " + future.isCompletedExceptionally());


lianetm · 2024-09-23T20:10:52Z

core/src/test/scala/integration/kafka/api/PlaintextConsumerSubscriptionTest.scala

@@ -244,6 +244,7 @@ class PlaintextConsumerSubscriptionTest extends AbstractConsumerTest {
    }, waitTimeMs = 5000, msg = "An InvalidTopicException should be thrown.")

    assertEquals(s"Invalid topics: [${invalidTopicName}]", exception.getMessage)
+    Thread.sleep(1000)


lianetm · 2024-09-23T20:28:47Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

@@ -524,7 +543,7 @@ public void transitionToJoining() {
     * to leave the group has been sent out.
     */
    public CompletableFuture<Void> leaveGroup() {
-        if (isNotInGroup()) {
+        if (isNotInGroup() || state == MemberState.JOINING) {


I don't quite get how this change relates to this PR, why is it that we need this here?

Also, this is effectively ignoring the leave group (unsubscribe) if JOINING, which I would expect is not right (the member will remain JOINING and may even become STABLE, never leave). Before this change, JOINING + consumer.unsubscribe => LEAVING (member would run the full leave flow and attempt to send the leave HB). With this change, JOINING + consumer.unsubscribe => still JOINING. Is that the intention?

I know that the leave while joining has challenges to solve so that it can be processed correctly (KIP-1082), but I would expect that we keep the intention we had here, even after the KIP-1082 fixes: we should attempt to leave the group when there's a call to unsubscribe while waiting for the join response (JOINING) because the broker may have already processed the join.

FrankYang0529 · 2024-09-24T09:01:59Z

Hi @lianetm, I'm sorry that there was a testing commit. I've removed it. Also, I change shouldTransitionToJoining to hasSubscriptionUpdated and set hasSubscriptionUpdated to true in AbstractMembershipManager#hasSubscriptionUpdated without checking subscription state.

lianetm

Hey @FrankYang0529 , thanks for the updates. Just some minor comments left.

lianetm · 2024-09-24T19:04:06Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

+     * AtomicBoolean to track whether the subscription is updated.
+     * If it's true and subscription state is UNSUBSCRIBED, the next {@link #maybeJoinGroup()} will change member state to JOINING.
+     */
+    private final AtomicBoolean hasSubscriptionUpdated = new AtomicBoolean(false);


nit: what about simply subscriptionUpdated ...to track whether the subscription has been updated

Yes, update the variable and function to subscriptionUpdated. Thank you.

lianetm · 2024-09-24T19:16:03Z

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

@@ -1617,10 +1617,25 @@ public void testRevokePartitionsUsesTopicNamesLocalCacheWhenMetadataNotAvailable
    }

    @Test
-    public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+    public void testOnSubscriptionUpdatedDoesNotSetShouldTransitionToJoiningIfInGroup() {


there is no "shouldTransitionToJoining" var/concept anymore, so should we update this ? maybe simply ...DoesNotTransitionToJoiningIfInGroup

Updated the test case name to testOnSubscriptionUpdatedDoesNotTransitionToJoiningIfInGroup.

lianetm · 2024-09-24T19:16:36Z

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

+    }
+
+    @Test
+    public void testOnSubscriptionUpdatedSetShouldTransitionToJoiningIfNotInGroup() {


testOnSubscriptionUpdatedTransitionsToJoiningIOnPollfNotInGroup?

Changed the test case name to testOnSubscriptionUpdatedTransitionsToJoiningOnPollIfNotInGroup. Thanks.

lianetm · 2024-09-24T19:18:07Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/ShareMembershipManagerTest.java

@@ -1137,10 +1137,25 @@ public void testRevokePartitionsUsesTopicNamesLocalCacheWhenMetadataNotAvailable
    }

    @Test
-    public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+    public void testOnSubscriptionUpdatedDoesNotSetShouldTransitionToJoiningIfInGroup() {


let's make sure to align these with the names in the consumer tests if they change with the comments above

Yes, align test case names between ConsumerMembershipManagerTest and ShareMembershipManagerTest.

lianetm · 2024-09-24T19:24:46Z

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

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


I would say this test only makes sense withGroupId=true right? if there is no groupId, the processor will have null HBMgr, MembershipMgr (there cannot be a SubscriptionChange event without a groupId, the api call to subscribe requires groupId)

Yes, I leave withGroupId=true case here. Thank you.

lianetm

Hey @FrankYang0529 , could you please update the PR description where it says:

When calling AsyncKafkaConsumer#subscribe, send the JoinRequest if the flag is true.

I guess you meant when calling poll transition to joining if the subscription has been updated?

With that (and the last nit up to you) this looks good to me. Thanks!

lianetm · 2024-09-26T10:53:37Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java

+     * from the {@link #onSubscriptionUpdated} to fulfill the requirement of the "rebalances will only occur during an
+     * active call to {@link org.apache.kafka.clients.consumer.KafkaConsumer#poll(Duration)}"
+     */
+    public void maybeJoinGroup() {


nit: what do you think about calling this something like onConsumerPoll? Just to make the func self explanatory (on poll -> transition to joining), no need to find usages to understand what this is all about. No strong feeling though, up to you.

Thanks for the suggestion. I updated PR description and change maybeJoinGroup to onConsumerPoll.

…consumer.poll Signed-off-by: PoAn Yang <[email protected]>

lianetm

Thanks @FrankYang0529! LGTM.

lianetm · 2024-09-26T14:30:20Z

FYI, I filed https://issues.apache.org/jira/browse/KAFKA-17623 for a flaky consumer integration test that I noticed here, has been flaky for a while.

FrankYang0529 · 2024-09-26T14:38:31Z

FYI, I filed https://issues.apache.org/jira/browse/KAFKA-17623 for a flaky consumer integration test that I noticed here, has been flaky for a while.

Okay. I will take a look. Thanks for filing the issue.

chia7712 · 2024-09-26T19:22:38Z

This PR causes the following build error:

/home/chia7712/project/kafka/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:3500: error: method requestGenerated(MockClient,ApiKeys) is already defined in class KafkaConsumerTest
    private boolean requestGenerated(MockClient client, ApiKeys apiKey) {

lianetm · 2024-09-26T19:24:44Z

uhm builds completed successfully but I guess it conflicted with a recent merge. Checking now and will open PR right away. Thanks @chia7712 !

lianetm · 2024-09-26T19:39:47Z

Removing dup method here #17291

mumrah · 2024-09-26T20:13:44Z

Once ASF signs off on the merge queue, we can completely avoid this class of failure on trunk. According to Infra team, we are a month or two away from being able to enable it.

lianetm · 2024-09-26T20:23:32Z

That will be helpful, it was really my miss not asking to have this rebased one last time.

…consumer.poll (apache#17165) Reviewers: Lianet Magrans <[email protected]>, TaiJuWu <[email protected]>

FrankYang0529 force-pushed the KAFKA-17154 branch from df12479 to 42ac730 Compare September 12, 2024 12:43

TaiJuWu reviewed Sep 13, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17154 branch from 42ac730 to 23ee599 Compare September 13, 2024 11:10

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

FrankYang0529 force-pushed the KAFKA-17154 branch from 23ee599 to 7b83257 Compare September 17, 2024 02:19

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

FrankYang0529 force-pushed the KAFKA-17154 branch 2 times, most recently from 91f5a98 to 804f17c Compare September 17, 2024 04:06

lianetm reviewed Sep 17, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17154 branch 2 times, most recently from a01febd to 23d386f Compare September 18, 2024 15:38

FrankYang0529 marked this pull request as draft September 19, 2024 12:23

FrankYang0529 force-pushed the KAFKA-17154 branch from 23d386f to 1cebf23 Compare September 22, 2024 05:57

github-actions bot added core Kafka Broker clients labels Sep 22, 2024

FrankYang0529 marked this pull request as ready for review September 23, 2024 11:55

lianetm reviewed Sep 23, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17154 branch from 1cebf23 to 29f1f4f Compare September 24, 2024 08:44

FrankYang0529 force-pushed the KAFKA-17154 branch from 29f1f4f to 217df42 Compare September 24, 2024 09:06

lianetm reviewed Sep 24, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-17154 branch from 217df42 to a091698 Compare September 25, 2024 08:45

lianetm reviewed Sep 26, 2024

View reviewed changes

KAFKA-17154: New consumer subscribe may join group without a call to …

87575f2

…consumer.poll Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 force-pushed the KAFKA-17154 branch from a091698 to 87575f2 Compare September 26, 2024 11:53

lianetm approved these changes Sep 26, 2024

View reviewed changes

lianetm removed the core Kafka Broker label Sep 26, 2024

lianetm merged commit 4e52a12 into apache:trunk Sep 26, 2024
7 of 9 checks passed

chia7712 mentioned this pull request Sep 26, 2024

KAFKA-17628 New workflows for automating run approvals #17290

Merged

FrankYang0529 deleted the KAFKA-17154 branch September 27, 2024 00:16

bbejeck pushed a commit to bbejeck/kafka that referenced this pull request Sep 28, 2024

KAFKA-17154: New consumer subscribe may join group without a call to …

832fb14

…consumer.poll (apache#17165) Reviewers: Lianet Magrans <[email protected]>, TaiJuWu <[email protected]>

		membershipManager.onSubscriptionUpdated();
		assertFalse(membershipManager.subscriptionUpdated());

		@@ -189,6 +191,9 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl

		private final Time time;

		// AtomicBoolean to track if the subscription has been updated

		CompletableFuture<Void> future = requestManagers.consumerHeartbeatRequestManager.get().membershipManager().leaveGroup();
		System.out.println("UnsubscribeEvent: " + future.isCompletedExceptionally());

KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll #17165

KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll #17165

Conversation

FrankYang0529 commented Sep 11, 2024 • edited Loading

Committer Checklist (excluded from commit message)

TaiJuWu commented Sep 12, 2024 • edited Loading

FrankYang0529 commented Sep 12, 2024

TaiJuWu commented Sep 13, 2024

Choose a reason for hiding this comment

FrankYang0529 Sep 17, 2024 • edited Loading

Choose a reason for hiding this comment

TaiJuWu Sep 17, 2024 • edited Loading

Choose a reason for hiding this comment

FrankYang0529 Sep 17, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lianetm commented Sep 16, 2024

FrankYang0529 commented Sep 17, 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

FrankYang0529 commented Sep 19, 2024

lianetm commented Sep 19, 2024

FrankYang0529 commented Sep 19, 2024

lianetm commented Sep 23, 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

FrankYang0529 commented Sep 24, 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

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

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

lianetm left a comment

Choose a reason for hiding this comment

lianetm commented Sep 26, 2024

FrankYang0529 commented Sep 26, 2024

chia7712 commented Sep 26, 2024

lianetm commented Sep 26, 2024

lianetm commented Sep 26, 2024

mumrah commented Sep 26, 2024 • edited Loading

lianetm commented Sep 26, 2024

FrankYang0529 commented Sep 11, 2024 •

edited

Loading

TaiJuWu commented Sep 12, 2024 •

edited

Loading

FrankYang0529 Sep 17, 2024 •

edited

Loading

TaiJuWu Sep 17, 2024 •

edited

Loading

FrankYang0529 Sep 17, 2024 •

edited

Loading

lianetm Sep 24, 2024 •

edited

Loading

lianetm left a comment •

edited

Loading

mumrah commented Sep 26, 2024 •

edited

Loading