dajac commented on code in PR #14857:
URL: https://github.com/apache/kafka/pull/14857#discussion_r1409010440
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -223,8 +232,65 @@ public void testFencingWhenStateIsReconciling() {
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
}
+ @Test
+ public void testFencingWhenStateIsPrepareLeaving() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+
+ // Start leaving group, blocked waiting for commit of all consumed to
complete.
+ CompletableFuture<Void> commitResult =
mockPrepareLeavingStuckCommitting();
Review Comment:
nit: There is an extra space before `commitResult`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -531,7 +537,13 @@ private CompletableFuture<Void>
invokeOnPartitionsRevokedOrLostToReleaseAssignme
* request is sent out with it.
*/
private void transitionToSendingLeaveGroup() {
- memberEpoch = ConsumerGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
+ if (state == MemberState.FATAL) {
+ log.warn("Member won't send leave group request because it is in
FATAL state");
Review Comment:
nit: `Member {} with epoch {}`? I am not sure whether the epoch makes any
sense here.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -223,8 +232,65 @@ public void testFencingWhenStateIsReconciling() {
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
}
+ @Test
+ public void testFencingWhenStateIsPrepareLeaving() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+
+ // Start leaving group, blocked waiting for commit of all consumed to
complete.
+ CompletableFuture<Void> commitResult =
mockPrepareLeavingStuckCommitting();
+ membershipManager.leaveGroup();
+ assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
+
+ // Get fenced while preparing to leave the group. Member should not
try to rejoin and
+ // continue leaving the group as it was before getting fenced.
+ mockMemberHasAutoAssignedPartition();
+ membershipManager.transitionToFenced();
Review Comment:
Is there a way to verify that we don't (re-)trigger the lost callback here?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -451,6 +517,41 @@ public void testFatalFailureWhenStateIsStable() {
testStateUpdateOnFatalFailure(membershipManager);
}
+ @Test
+ public void testFatalFailureWhenStateIsPrepareLeaving() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+
+ // Start leaving group, blocked waiting for commit of all consumed to
complete.
+ CompletableFuture<Void> commitResult =
mockPrepareLeavingStuckCommitting();
Review Comment:
nit: There is an extra space here as well.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java:
##########
@@ -365,6 +366,11 @@ private void replaceUnresolvedAssignmentWithNewAssignment(
*/
@Override
public void transitionToFenced() {
+ if (state == MemberState.PREPARE_LEAVING || state ==
MemberState.LEAVING) {
+ log.debug("Member ID {} got fenced but it is already leaving the
group with state {}," +
Review Comment:
nit: `Member {} with epoch {}`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java:
##########
@@ -223,8 +232,65 @@ public void testFencingWhenStateIsReconciling() {
verify(subscriptionState).assignFromSubscribed(Collections.emptySet());
}
+ @Test
+ public void testFencingWhenStateIsPrepareLeaving() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+
+ // Start leaving group, blocked waiting for commit of all consumed to
complete.
+ CompletableFuture<Void> commitResult =
mockPrepareLeavingStuckCommitting();
+ membershipManager.leaveGroup();
+ assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
+
+ // Get fenced while preparing to leave the group. Member should not
try to rejoin and
+ // continue leaving the group as it was before getting fenced.
+ mockMemberHasAutoAssignedPartition();
+ membershipManager.transitionToFenced();
+ assertEquals(MemberState.PREPARE_LEAVING, membershipManager.state());
+ assertNotEquals(0, membershipManager.memberEpoch());
+
+ // When commit completes member should transition to LEAVE.
+ commitResult.complete(null);
+ assertEquals(MemberState.LEAVING, membershipManager.state());
+ }
+
+ @Test
+ public void testFencingWhenStateIsLeaving() {
+ MembershipManagerImpl membershipManager = createMemberInStableState();
+
+ // Start leaving group.
+ mockLeaveGroup();
+ membershipManager.leaveGroup();
+ assertEquals(MemberState.LEAVING, membershipManager.state());
+
+ // Get fenced while leaving. Member should not try to rejoin and
continue leaving the
+ // group as it was before getting fenced.
+ mockMemberHasAutoAssignedPartition();
+ membershipManager.transitionToFenced();
+ assertEquals(MemberState.LEAVING, membershipManager.state());
Review Comment:
If the member got removed from the group, the final HB will fail because the
member does not exist anymore. Do we handle this case?
--
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
