aiquestion commented on code in PR #12140:
URL: https://github.com/apache/kafka/pull/12140#discussion_r895136385
##########
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:
can't think of any name for this 2000 timeout. so i just changed it to
rebalance timtoue. does that make sense? :-p
##########
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:
can't think of any name for this 2000 timeout. so i just changed it to
rebalance timeout. does that make sense? :-p
--
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
