kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [kafka] branch 2.1 updated: KAFKA-8104: Consumer cannot rejoin to the group after rebalancing (#7460)
Date Fri, 29 May 2020 22:25:08 GMT
This is an automated email from the ASF dual-hosted git repository.

jgus pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.1 by this push:
     new 47c70a7  KAFKA-8104: Consumer cannot rejoin to the group after rebalancing (#7460)
47c70a7 is described below

commit 47c70a7416f4ed5cbf204485b7d9c391f5e82b7c
Author: Nikolay <nizhikov@apache.org>
AuthorDate: Thu Oct 17 08:53:14 2019 +0300

    KAFKA-8104: Consumer cannot rejoin to the group after rebalancing (#7460)
    
    This PR contains the fix of race condition bug between "consumer thread" and "consumer
coordinator heartbeat thread". It reproduces in many production environments.
    
    Condition for reproducing:
    
    1. Consumer thread initiates rejoin to the group because of commit timeout. Call of AbstractCoordinator#joinGroupIfNeeded
which leads to sendJoinGroupRequest.
    2. JoinGroupResponseHandler writes to the AbstractCoordinator.this.generation new generation
data and leaves the synchronized section.
    3. Heartbeat thread executes mabeLeaveGroup and clears generation data via resetGenerationOnLeaveGroup.
    4. Consumer thread executes onJoinComplete(generation.generationId, generation.memberId,
generation.protocol, memberAssignment); with the cleared generation data. This leads to the
corresponding exception.
    
    Note this PR is a backport of #7460 for 2.1. It also contains part of #7451.
    
    Reviewers: Guozhang Wang <wangguoz@gmail.com>
    
    Co-authored-by: Nikolay <nizhikov@apache.org>
---
 .../consumer/internals/AbstractCoordinator.java    | 62 ++++++++++++++++------
 .../consumer/internals/ConsumerCoordinator.java    | 19 +++----
 .../internals/ConsumerCoordinatorTest.java         | 59 +++++++++++++++++++-
 .../runtime/distributed/WorkerCoordinator.java     |  2 +-
 4 files changed, 115 insertions(+), 27 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 d983087..0a90144 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
@@ -401,14 +401,32 @@ public abstract class AbstractCoordinator implements Closeable {
             }
 
             if (future.succeeded()) {
-                // Duplicate the buffer in case `onJoinComplete` does not complete and needs
to be retried.
-                ByteBuffer memberAssignment = future.value().duplicate();
-                onJoinComplete(generation.generationId, generation.memberId, generation.protocol,
memberAssignment);
+                Generation generationSnapshot;
+
+                // Generation data maybe concurrently cleared by Heartbeat thread.
+                // Can't use synchronized for {@code onJoinComplete}, because it can be long
enough
+                // and  shouldn't block hearbeat thread.
+                // See {@link PlaintextConsumerTest#testMaxPollIntervalMsDelayInAssignment
+                synchronized (this) {
+                    generationSnapshot = this.generation;
+                }
 
-                // We reset the join group future only after the completion callback returns.
This ensures
-                // that if the callback is woken up, we will retry it on the next joinGroupIfNeeded.
-                resetJoinGroupFuture();
-                needsJoinPrepare = true;
+                if (generationSnapshot != Generation.NO_GENERATION) {
+                    // Duplicate the buffer in case `onJoinComplete` does not complete and
needs to be retried.
+                    ByteBuffer memberAssignment = future.value().duplicate();
+
+                    onJoinComplete(generationSnapshot.generationId, generationSnapshot.memberId,
generationSnapshot.protocol, memberAssignment);
+
+                    // We reset the join group future only after the completion callback
returns. This ensures
+                    // that if the callback is woken up, we will retry it on the next joinGroupIfNeeded.
+                    resetJoinGroupFuture();
+                    needsJoinPrepare = true;
+                } else {
+                    log.info("Generation data was cleared by heartbeat thread. Initiating
rejoin.");
+                    resetStateAndRejoin();
+
+                    return false;
+                }
             } else {
                 resetJoinGroupFuture();
                 final RuntimeException exception = future.exception();
@@ -429,6 +447,11 @@ public abstract class AbstractCoordinator implements Closeable {
         this.joinFuture = null;
     }
 
+    private void resetStateAndRejoin() {
+        rejoinNeeded = true;
+        state = MemberState.UNJOINED;
+    }
+
     private synchronized RequestFuture<ByteBuffer> initiateJoinGroup() {
         // we store the join future in case we are woken up by the user after beginning the
         // rebalance in the call to poll below. This ensures that we do not mistakenly attempt
@@ -447,12 +470,17 @@ public abstract class AbstractCoordinator implements Closeable {
                     // handle join completion in the callback so that the callback will be
invoked
                     // even if the consumer is woken up before finishing the rebalance
                     synchronized (AbstractCoordinator.this) {
-                        log.info("Successfully joined group with generation {}", generation.generationId);
-                        state = MemberState.STABLE;
-                        rejoinNeeded = false;
+                        if (generation != Generation.NO_GENERATION) {
+                            log.info("Successfully joined group with generation {}", generation.generationId);
+                            state = MemberState.STABLE;
+                            rejoinNeeded = false;
 
-                        if (heartbeatThread != null)
-                            heartbeatThread.enable();
+                            if (heartbeatThread != null)
+                                heartbeatThread.enable();
+                        } else {
+                            log.info("Generation data was cleared by heartbeat thread. Rejoin
failed.");
+                            state = MemberState.UNJOINED;
+                        }
                     }
                 }
 
@@ -722,19 +750,23 @@ public abstract class AbstractCoordinator implements Closeable {
      * Get the current generation state if the group is stable.
      * @return the current generation or null if the group is unjoined/rebalancing
      */
-    protected synchronized Generation generation() {
+    protected synchronized Generation generationIfStable() {
         if (this.state != MemberState.STABLE)
             return null;
         return generation;
     }
 
+    // Visible for testing
+    protected synchronized Generation generation() {
+        return generation;
+    }
+
     /**
      * Reset the generation and memberId because we have fallen out of the group.
      */
     protected synchronized void resetGeneration() {
         this.generation = Generation.NO_GENERATION;
-        this.rejoinNeeded = true;
-        this.state = MemberState.UNJOINED;
+        resetStateAndRejoin();
     }
 
     protected synchronized void requestRejoin() {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index 729011c..dd73341 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -515,7 +515,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
                                                                         final Timer timer)
{
         if (partitions.isEmpty()) return Collections.emptyMap();
 
-        final Generation generation = generation();
+        final Generation generation = generationIfStable();
         if (pendingCommittedOffsetRequest != null && !pendingCommittedOffsetRequest.sameRequest(partitions,
generation)) {
             // if we were waiting for a different request, then just clear it.
             pendingCommittedOffsetRequest = null;
@@ -767,16 +767,17 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
         }
 
         final Generation generation;
-        if (subscriptions.partitionsAutoAssigned())
-            generation = generation();
-        else
+        if (subscriptions.partitionsAutoAssigned()) {
+            generation = generationIfStable();
+            // if the generation is null, we are not part of an active group (and we expect
to be).
+            // the only thing we can do is fail the commit and let the user rejoin the group
in poll()
+            if (generation == null) {
+                log.info("Failing OffsetCommit request since the consumer is not part of
an active group");
+                return RequestFuture.failure(new CommitFailedException());
+            }
+        } else
             generation = Generation.NO_GENERATION;
 
-        // if the generation is null, we are not part of an active group (and we expect to
be).
-        // the only thing we can do is fail the commit and let the user rejoin the group
in poll()
-        if (generation == null)
-            return RequestFuture.failure(new CommitFailedException());
-
         OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder(this.groupId,
offsetData).
                 setGenerationId(generation.generationId).
                 setMemberId(generation.memberId);
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
index af073f1..d091e1f 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java
@@ -41,6 +41,7 @@ import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.requests.AbstractRequest;
@@ -741,7 +742,7 @@ public class ConsumerCoordinatorTest {
         coordinator.maybeLeaveGroup();
         assertTrue(received.get());
 
-        AbstractCoordinator.Generation generation = coordinator.generation();
+        AbstractCoordinator.Generation generation = coordinator.generationIfStable();
         assertNull(generation);
     }
 
@@ -1905,7 +1906,7 @@ public class ConsumerCoordinatorTest {
     @Test
     public void testAutoCommitAfterCoordinatorBackToService() {
         ConsumerCoordinator coordinator = buildCoordinator(new Metrics(), assignors,
-                ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true, true);
+            ConsumerConfig.DEFAULT_EXCLUDE_INTERNAL_TOPICS, true, true);
         subscriptions.assignFromUser(Collections.singleton(t1p));
         subscriptions.seek(t1p, 100L);
 
@@ -1921,6 +1922,60 @@ public class ConsumerCoordinatorTest {
         assertEquals(100L, subscriptions.position(t1p).longValue());
     }
 
+    @Test
+    public void testConsumerRejoinAfterRebalance() throws Exception {
+        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, false,
false)) {
+            coordinator.ensureActiveGroup();
+
+            prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS);
+
+            try {
+                coordinator.commitOffsetsSync(singletonMap(t1p, new OffsetAndMetadata(100L)),
+                    time.timer(Long.MAX_VALUE));
+                fail("Expected commit failure");
+            } catch (CommitFailedException e) {
+                // Expected
+            }
+
+            assertFalse(client.hasPendingResponses());
+            assertFalse(client.hasInFlightRequests());
+
+            int generationId = 42;
+            String memberId = "consumer-42";
+
+            client.prepareResponse(joinGroupFollowerResponse(generationId, memberId, "leader",
Errors.NONE));
+
+            MockTime time = new MockTime(1);
+
+            //onJoinPrepare will be executed and onJoinComplete will not.
+            boolean res = coordinator.joinGroupIfNeeded(time.timer(2));
+
+            assertFalse(res);
+            assertFalse(client.hasPendingResponses());
+            //SynGroupRequest not responded.
+            assertEquals(1, client.inFlightRequestCount());
+            assertEquals(generationId, coordinator.generation().generationId);
+            assertEquals(memberId, coordinator.generation().memberId);
+
+            // Imitating heartbeat thread that clears generation data.
+            coordinator.maybeLeaveGroup();
+
+            assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation());
+
+            client.respond(syncGroupResponse(singletonList(t1p), Errors.NONE));
+
+            //Join future should succeed but generation already cleared so result of join
is false.
+            res = coordinator.joinGroupIfNeeded(time.timer(1));
+
+            assertFalse(res);
+            assertFalse(client.hasPendingResponses());
+
+            // We just have the LeaveGroup
+            assertEquals(1, client.inFlightRequestCount());
+            assertEquals(ApiKeys.LEAVE_GROUP, client.requests().peek().apiKey());
+        }
+    }
+
     private ConsumerCoordinator prepareCoordinatorForCloseTest(final boolean useGroupManagement,
                                                                final boolean autoCommit,
                                                                final boolean leaveGroup)
{
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
index 103a323..5a83435 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
@@ -295,7 +295,7 @@ public final class WorkerCoordinator extends AbstractCoordinator implements
Clos
     }
 
     public String memberId() {
-        Generation generation = generation();
+        Generation generation = generationIfStable();
         if (generation != null)
             return generation.memberId;
         return JoinGroupRequest.UNKNOWN_MEMBER_ID;


Mime
View raw message