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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
