(kafka) branch 3.7 updated: MINOR: Client state machine fix for transition to stable on initial empty assignment (#15033)

[ca...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

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


The following commit(s) were added to refs/heads/3.7 by this push:
     new 89de4219212 MINOR: Client state machine fix for transition to stable on initial empty assignment (#15033)
89de4219212 is described below

commit 89de4219212455da5313e645d2729fe3602578e8
Author: Lianet Magrans <98...@users.noreply.github.com>
AuthorDate: Wed Dec 20 03:22:54 2023 -0500

    MINOR: Client state machine fix for transition to stable on initial empty assignment (#15033)
    
    This includes a fix for ensuring that a member transitions to STABLE when joining a group, in the case where it receives an initial empty assignment.
    
    Reviewer: Bruno Cadonna <ca...@apache.org>
---
 .../kafka/clients/consumer/internals/MembershipManagerImpl.java   | 6 ++----
 .../clients/consumer/internals/MembershipManagerImplTest.java     | 8 ++++++++
 2 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
index 1cdd26977d0..21211ae4a2b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImpl.java
@@ -405,10 +405,8 @@ public class MembershipManagerImpl implements MembershipManager, ClusterResource
                 assignmentUnresolved, currentAssignment);
             // Make sure we transition the member back to STABLE if it was RECONCILING (ex.
             // member was RECONCILING unresolved assignments that were just removed by the
-            // broker).
-            if (state == MemberState.RECONCILING) {
-                // This is the case where a member was RECONCILING an unresolved
-                // assignment that was removed by the broker in a following assignment.
+            // broker), or JOINING (member joining received empty assignment).
+            if (state == MemberState.RECONCILING || state == MemberState.JOINING) {
                 transitionTo(MemberState.STABLE);
             }
         }
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
index 88058d1d541..15fbd4b7220 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/MembershipManagerImplTest.java
@@ -1403,6 +1403,14 @@ public class MembershipManagerImplTest {
         assertEquals(MemberState.JOINING, membershipManager.state());
     }
 
+    @Test
+    public void testMemberJoiningTransitionsToStableWhenReceivingEmptyAssignment() {
+        MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup(null);
+        assertEquals(MemberState.JOINING, membershipManager.state());
+        receiveEmptyAssignment(membershipManager);
+        assertEquals(MemberState.STABLE, membershipManager.state());
+    }
+
     private MembershipManagerImpl mockMemberSuccessfullyReceivesAndAcksAssignment(
             Uuid topicId, String topicName, List<Integer> partitions) {
         MembershipManagerImpl membershipManager = createMembershipManagerJoiningGroup();