(kafka) branch trunk 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 trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


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

commit db8af514264491e52d737300bf703d473e3c676a
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 62fe07fd32c..2de533eaa2a 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 a75b613a17a..f299e4ebe34 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 void dropAssignedPartitions() {
         SortedSet<TopicPartition> droppedPartitions = new TreeSet<>(MembershipManagerImpl.TOPIC_PARTITION_COMPARATOR);
         droppedPartitions.addAll(subscriptionState.assignedPartitions());