You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/09/11 14:46:23 UTC
[PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
FrankYang0529 opened a new pull request, #17165:
URL: https://github.com/apache/kafka/pull/17165
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#subscribe`, 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)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1774824100
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -189,6 +191,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
private final Time time;
+ /**
+ * 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);
Review Comment:
Yes, update the variable and function to `subscriptionUpdated`. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1773920760
##########
clients/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() {
Review Comment:
let's make sure to align these with the names in the consumer tests if they change with the comments above
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1772011078
##########
clients/src/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);
Review Comment:
lets remove this please
##########
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)
Review Comment:
remove?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -466,6 +474,17 @@ String memberIdInfoForLog() {
*/
public void onSubscriptionUpdated() {
if (state == MemberState.UNSUBSCRIBED) {
Review Comment:
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?
##########
clients/src/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);
CompletableFuture<Void> future = requestManagers.consumerHeartbeatRequestManager.get().membershipManager().leaveGroup();
+ System.out.println("UnsubscribeEvent: " + future.isCompletedExceptionally());
Review Comment:
ditto
##########
clients/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) {
Review Comment:
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1774832572
##########
clients/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() {
Review Comment:
Yes, align test case names between `ConsumerMembershipManagerTest` and `ShareMembershipManagerTest`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1776831516
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -458,14 +466,24 @@ String memberIdInfoForLog() {
}
/**
- * Join the group with the updated subscription, if the member is not part of it yet. If the
- * member is already part of the group, this will only ensure that the updated subscription
+ * Set {@link #subscriptionUpdated} to true to indicate that the subscription has been updated.
+ * The next {@link #maybeJoinGroup()} will join the group with the updated subscription, if the member is not part of it yet.
+ * If the member is already part of the group, this will only ensure that the updated subscription
* is included in the next heartbeat request.
* <p/>
* Note that list of topics of the subscription is taken from the shared subscription state.
*/
public void onSubscriptionUpdated() {
- if (state == MemberState.UNSUBSCRIBED) {
+ subscriptionUpdated.compareAndSet(false, true);
+ }
+
+ /**
+ * Join the group if the member is not part of it yet. This function separates {@link #transitionToJoining}
+ * 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() {
Review Comment:
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2353390179
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!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1774825467
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertTrue(membershipManager.hasSubscriptionUpdated());
+ membershipManager.maybeJoinGroup();
verify(membershipManager, never()).transitionToJoining();
+ assertFalse(membershipManager.hasSubscriptionUpdated());
+ }
+
+ @Test
+ public void testOnSubscriptionUpdatedSetShouldTransitionToJoiningIfNotInGroup() {
Review Comment:
Changed the test case name to `testOnSubscriptionUpdatedTransitionsToJoiningOnPollIfNotInGroup`. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1774833062
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -208,6 +207,35 @@ public void testSeekUnvalidatedEventWithException() {
assertInstanceOf(IllegalStateException.class, e.getCause());
}
+ @Test
+ public void testPollEvent() {
+ PollEvent event = new PollEvent(12345);
+
+ setupProcessor(true);
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ processor.process(event);
+ verify(commitRequestManager).updateAutoCommitTimer(12345);
+ verify(membershipManager).maybeJoinGroup();
+ verify(heartbeatRequestManager).resetPollTimer(12345);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSubscriptionChangeEvent(boolean withGroupId) {
Review Comment:
Yes, I leave `withGroupId=true` case here. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377143597
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762256492
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762181434
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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 about your idea? Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377758987
uhm builds completed successfully but I guess it conflicted with a recent merge. Checking now and will open PR right away. Thanks @chia7712 !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377788086
Removing dup method here https://github.com/apache/kafka/pull/17291
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1776908662
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -458,14 +466,24 @@ String memberIdInfoForLog() {
}
/**
- * Join the group with the updated subscription, if the member is not part of it yet. If the
- * member is already part of the group, this will only ensure that the updated subscription
+ * Set {@link #subscriptionUpdated} to true to indicate that the subscription has been updated.
+ * The next {@link #maybeJoinGroup()} will join the group with the updated subscription, if the member is not part of it yet.
+ * If the member is already part of the group, this will only ensure that the updated subscription
* is included in the next heartbeat request.
* <p/>
* Note that list of topics of the subscription is taken from the shared subscription state.
*/
public void onSubscriptionUpdated() {
- if (state == MemberState.UNSUBSCRIBED) {
+ subscriptionUpdated.compareAndSet(false, true);
+ }
+
+ /**
+ * Join the group if the member is not part of it yet. This function separates {@link #transitionToJoining}
+ * 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() {
Review Comment:
Thanks for the suggestion. I updated PR description and change `maybeJoinGroup` to `onConsumerPoll`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Sorry for unclear.
Make `application` sleep here is just for making `network` have time to send `join request` without poll in original code.
It can help to reproduce this bug easier and more stable, the test can pass sometime without it.
In order to make sure this behavior which we promise in docs, two operation should not too close in timeline. Does it make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2360907781
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #17165:
URL: https://github.com/apache/kafka/pull/17165
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2370690086
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1763830133
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertFalse(membershipManager.subscriptionUpdated());
Review Comment:
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)
##########
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 {
Review Comment:
typo testPoll**Sends**RequestForRebalance? (and maybe clearer `testPollSendsRequestToJoin`?)
##########
clients/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
Review Comment:
this is truly more to know if we should join after the subscription has been updated right? (true if not in group)
##########
clients/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() {
Review Comment:
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)
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertFalse(membershipManager.subscriptionUpdated());
+ membershipManager.maybeJoinGroup();
verify(membershipManager, never()).transitionToJoining();
}
+ @Test
+ public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+ ConsumerMembershipManager membershipManager = createMembershipManager(null);
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ membershipManager.onSubscriptionUpdated();
Review Comment:
should we verify here that there was no `transitionToJoining`? (to be sure that the transition is only triggered after the call to `maybeJoinGroup`)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1757861442
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Maybe we need to make `application thread`sleep here to make sure `network thread` does not send join request.
WDYT?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762256492
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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 `ConsumerMembershipManager` doesn't call `maybeJoinGroup`.
In `ConsumerMembershipManagerTest#testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup`, we check `ConsumerMembershipManager#onSubscriptionUpdated` keeps member state as `UNSUBSCRIBED` and calling `transitionToJoining` in `maybeJoinGroup`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Sorry for unclear.
Make `application` sleep here is just for making `network` have time to send `join request` without poll in original code.
It can help to reproduce this bug easier and more stable, the test can pass sometime without it.
In order to make sure this behavior which we promise in docs, two operation (subscribe and poll) should not too close in timeline. Does it make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Sorry for the confusion.
The reason for making the application sleep here is to give the network enough time to send the join request. 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. Does that make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1765288070
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertFalse(membershipManager.subscriptionUpdated());
Review Comment:
Change variable from `subscriptionUpdated` to `shouldTransitionToJoining` and add more comments for it. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1773915749
##########
clients/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() {
Review Comment:
there is no "shouldTransitionToJoining" var/concept anymore, so should we update this ? maybe simply `...DoesNotTransitionToJoiningIfInGroup`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractMembershipManager.java:
##########
@@ -189,6 +191,12 @@ public abstract class AbstractMembershipManager<R extends AbstractResponse> impl
private final Time time;
+ /**
+ * 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);
Review Comment:
nit: what about simply `subscriptionUpdated` ...to track whether the subscription has been updated
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertTrue(membershipManager.hasSubscriptionUpdated());
+ membershipManager.maybeJoinGroup();
verify(membershipManager, never()).transitionToJoining();
+ assertFalse(membershipManager.hasSubscriptionUpdated());
+ }
+
+ @Test
+ public void testOnSubscriptionUpdatedSetShouldTransitionToJoiningIfNotInGroup() {
Review Comment:
`testOnSubscriptionUpdatedTransitionsToJoiningIOnPollfNotInGroup`?
##########
clients/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() {
Review Comment:
let's sure make sure to align these with the names in the consumer tests if they change with the comments above
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -208,6 +207,35 @@ public void testSeekUnvalidatedEventWithException() {
assertInstanceOf(IllegalStateException.class, e.getCause());
}
+ @Test
+ public void testPollEvent() {
+ PollEvent event = new PollEvent(12345);
+
+ setupProcessor(true);
+ when(heartbeatRequestManager.membershipManager()).thenReturn(membershipManager);
+ processor.process(event);
+ verify(commitRequestManager).updateAutoCommitTimer(12345);
+ verify(membershipManager).maybeJoinGroup();
+ verify(heartbeatRequestManager).resetPollTimer(12345);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSubscriptionChangeEvent(boolean withGroupId) {
Review Comment:
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)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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. Does that make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2354390112
Hi @lianetm, yes, the PR is ready for review. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762256492
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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(ApplicationEvent)` doesn't call `ConsumerMembershipManager#maybeJoinGroup`.
In `ConsumerMembershipManagerTest#testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup`, we check `ConsumerMembershipManager#onSubscriptionUpdated` keeps member state as `UNSUBSCRIBED` and calling `transitionToJoining` in `maybeJoinGroup`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1774824804
##########
clients/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() {
Review Comment:
Updated the test case name to `testOnSubscriptionUpdatedDoesNotTransitionToJoiningIfInGroup`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Sorry for unclear.
Make `application` sleep here is just for making `network` have time to send `join request` without poll in original code.
It can help to reproduce this bug easier and more stable, the test can pass sometime without it on local.
Also, in order to make sure this behavior which we promise in docs, two operation (subscribe and poll) should not too close in timeline. Does it make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762204798
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Sorry for unclear.
Make `application` sleep here is just for making `network` have time to send `join request` without poll in original code.
It can help to reproduce this bug easier and more stable, the test can pass sometime without it.
In order to make sure this behavior which we promise in docs, two operation (subscribe and poll) should not too close in timeline. Does it make sense?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762181434
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1765334799
##########
clients/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() {
ConsumerMembershipManager membershipManager = createMemberInStableState();
membershipManager.onSubscriptionUpdated();
+ assertFalse(membershipManager.subscriptionUpdated());
+ membershipManager.maybeJoinGroup();
verify(membershipManager, never()).transitionToJoining();
}
+ @Test
+ public void testOnSubscriptionUpdatedTransitionsToJoiningOnlyIfNotInGroup() {
+ ConsumerMembershipManager membershipManager = createMembershipManager(null);
+ assertEquals(MemberState.UNSUBSCRIBED, membershipManager.state());
+ membershipManager.onSubscriptionUpdated();
Review Comment:
Yes, added it. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2360845836
Hi @lianetm, I found that `PlaintextConsumerSubscriptionTest#testSubscribeInvalidTopic` is flaky in this PR, so I will convert to draft first. Thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377164009
> 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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2345877501
Hi @FrankYang0529 , I opened a PR for this behavior.
Could you take a look? https://github.com/apache/kafka/pull/17175
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2346186703
> 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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1762285675
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3443,6 +3443,33 @@ public void testPreventMultiThread(GroupProtocol groupProtocol) throws Interrupt
}
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testPollSendRequestForRebalance(GroupProtocol groupProtocol) throws InterruptedException {
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ initMetadata(client, Collections.singletonMap(topic, 1));
+ Node node = metadata.fetch().nodes().get(0);
+
+ 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));
Review Comment:
Thanks for your explanation.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2360922267
> 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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17165:
URL: https://github.com/apache/kafka/pull/17165#discussion_r1765301371
##########
clients/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() {
Review Comment:
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.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377848870
Once ASF signs off on the merge queue, we can completely avoid this class of trunk failure. According to Infra team, we are a month or two away from being able to enable it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377865786
That will be helpful, it was really my miss not asking to have this rebased one last time.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2377755306
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) {
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2367970018
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!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17154: New consumer subscribe may join group without a call to consumer.poll [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #17165:
URL: https://github.com/apache/kafka/pull/17165#issuecomment-2347750081
> > 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`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org