You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2022/11/05 12:22:57 UTC
[kafka] 03/03: Revert "KAFKA-13891: reset generation when syncgroup failed with REBALANCE_IN_PROGRESS (#12140)" (#12794)
This is an automated email from the ASF dual-hosted git repository.
showuon pushed a commit to branch 3.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 846f404a8ff1c7e96cbe093abe08c2561b18ebd2
Author: Shawn <wa...@bytedance.com>
AuthorDate: Sat Nov 5 14:05:48 2022 +0800
Revert "KAFKA-13891: reset generation when syncgroup failed with REBALANCE_IN_PROGRESS (#12140)" (#12794)
This reverts commit c23d60d56cd73118624192cf03f6e400f59027e7.
Reviewers: Luke Chen <sh...@gmail.com>
---
.../consumer/internals/AbstractCoordinator.java | 3 -
.../internals/AbstractCoordinatorTest.java | 65 ----------------------
2 files changed, 68 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index c78caaba059..80b515d601b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -825,9 +825,6 @@ public abstract class AbstractCoordinator implements Closeable {
} else if (error == Errors.REBALANCE_IN_PROGRESS) {
log.info("SyncGroup failed: The group began another rebalance. Need to re-join the group. " +
"Sent generation was {}", sentGeneration);
- // consumer didn't get assignment in this generation, so we need to reset generation
- // to avoid joinGroup with out-of-data ownedPartitions in cooperative rebalance
- resetStateOnResponseError(ApiKeys.SYNC_GROUP, error, false);
future.raise(error);
} else if (error == Errors.FENCED_INSTANCE_ID) {
// for sync-group request, even if the generation has changed we would not expect the instance id
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
index 4471b6f88cf..0745b99749f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
@@ -67,7 +67,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -504,54 +503,6 @@ public class AbstractCoordinatorTest {
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");
-
- rejoinedGeneration = 20;
- expectSyncGroup(rejoinedGeneration, memberId);
- mockClient.respond(joinGroupFollowerResponse(
- rejoinedGeneration,
- memberId,
- "leaderId",
- Errors.NONE,
- PROTOCOL_TYPE
- ));
- assertTrue(secondJoin.get());
- } finally {
- executor.shutdownNow();
- executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
- }
- }
-
@Test
public void testRejoinReason() {
setupCoordinator();
@@ -639,22 +590,6 @@ public class AbstractCoordinatorTest {
}, null, true);
}
- private void expectRebalanceInProgressForSyncGroup(
- int expectedGeneration,
- String expectedMemberId
- ) {
- mockClient.prepareResponse(body -> {
- if (!(body instanceof SyncGroupRequest)) {
- return false;
- }
- SyncGroupRequestData syncGroupRequest = ((SyncGroupRequest) body).data();
- return syncGroupRequest.generationId() == expectedGeneration
- && syncGroupRequest.memberId().equals(expectedMemberId)
- && syncGroupRequest.protocolType().equals(PROTOCOL_TYPE)
- && syncGroupRequest.protocolName().equals(PROTOCOL_NAME);
- }, syncGroupResponse(Errors.REBALANCE_IN_PROGRESS, PROTOCOL_TYPE, PROTOCOL_NAME));
- }
-
private void expectDisconnectInJoinGroup(
String expectedMemberId
) {