showuon commented on code in PR #12140:
URL: https://github.com/apache/kafka/pull/12140#discussion_r895147723
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java:
##########
@@ -488,6 +489,54 @@ public void testRetainMemberIdAfterSyncGroupDisconnect() {
ensureActiveGroup(rejoinedGeneration, memberId);
}
+ @Test
+ public void
testResetGenerationIdAfterSyncGroupFailedWithRebalanceInProgress() throws
InterruptedException, ExecutionException {
+ setupCoordinator();
+
+ String memberId = "memberId";
+ int generation = 5;
+
+ // Rebalance once to initialize the generation and memberId
+ mockClient.prepareResponse(groupCoordinatorResponse(node,
Errors.NONE));
+ expectJoinGroup("", generation, memberId);
+ expectSyncGroup(generation, memberId);
+ ensureActiveGroup(generation, memberId);
+
+ // Force a rebalance
+ coordinator.requestRejoin("Manual test trigger");
+ assertTrue(coordinator.rejoinNeededOrPending());
+
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ try {
+ // Return RebalanceInProgress in syncGroup
+ int rejoinedGeneration = 10;
+ expectJoinGroup(memberId, rejoinedGeneration, memberId);
+ expectRebalanceInProgressForSyncGroup(rejoinedGeneration,
memberId);
+ Future<Boolean> secondJoin = executor.submit(() ->
+
coordinator.ensureActiveGroup(mockTime.timer(Integer.MAX_VALUE)));
+
+ TestUtils.waitForCondition(() -> {
+ AbstractCoordinator.Generation currentGeneration =
coordinator.generation();
+ return currentGeneration.generationId ==
AbstractCoordinator.Generation.NO_GENERATION.generationId &&
+ currentGeneration.memberId.equals(memberId);
+ }, 2000, "Generation should be reset");
Review Comment:
@aiquestion , the current `REBALANCE_TIMEOUT_MS` is 60 seconds, which means
we'll wait 60 secs for generation reset. It's not correct. It should use 2
seconds as before. I think you can ignore my previous minor comment about
`2000` change, and revert the `REBALANCE_TIMEOUT_MS` back to `2000`. 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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
