(kafka) branch trunk updated: KAFKA-16251: Fix for not sending heartbeat while fenced (#15392)

[lucasbru]

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

lucasbru 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 2185004083e KAFKA-16251: Fix for not sending heartbeat while fenced 
(#15392)
2185004083e is described below

commit 2185004083ebb8f0b3a443132b5a33908c459c65
Author: Lianet Magrans <98415067+lian...@users.noreply.github.com>
AuthorDate: Fri Feb 23 04:56:05 2024 -0500

    KAFKA-16251: Fix for not sending heartbeat while fenced (#15392)
    
    Fix to ensure that a consumer that has been fenced by the coordinator stops 
sending heartbeats while it is on the FENCED state releasing its assignment 
(waiting for the onPartitionsLost callback to complete). Once the callback 
completes, the member transitions to JOINING and it's then when it should 
resume sending heartbeats again.
    
    Reviewers: Lucas Brutschy <lbruts...@confluent.io>
---
 .../consumer/internals/MembershipManagerImpl.java  | 14 +++++------
 .../internals/HeartbeatRequestManagerTest.java     | 27 ++++++++++++++++++++++
 .../internals/MembershipManagerImplTest.java       |  2 ++
 3 files changed, 35 insertions(+), 8 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 dd035506d4b..81e65dfd866 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
@@ -720,18 +720,16 @@ public class MembershipManagerImpl implements 
MembershipManager {
     }
 
     /**
-     * @return True if the member should not send heartbeats, which would be 
one of the following
-     * cases:
-     * <ul>
-     * <li>Member is not subscribed to any topics</li>
-     * <li>Member has received a fatal error in a previous heartbeat 
response</li>
-     * <li>Member is stale, meaning that it has left the group due to expired 
poll timer</li>
-     * </ul>
+     * @return True if the member should not send heartbeats, which is the 
case when it is in a
+     * state where it is not an active member of the group.
      */
     @Override
     public boolean shouldSkipHeartbeat() {
         MemberState state = state();
-        return state == MemberState.UNSUBSCRIBED || state == MemberState.FATAL 
|| state == MemberState.STALE;
+        return state == MemberState.UNSUBSCRIBED ||
+            state == MemberState.FATAL ||
+            state == MemberState.STALE ||
+            state == MemberState.FENCED;
     }
 
     /**
diff --git 
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
 
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
index 4016b74b27b..72a5c0349d2 100644
--- 
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
+++ 
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManagerTest.java
@@ -663,6 +663,33 @@ public class HeartbeatRequestManagerTest {
         assertEquals((double) randomSleepS, 
getMetric("last-heartbeat-seconds-ago").metricValue());
     }
 
+    @Test
+    public void 
testFencedMemberStopHeartbeatUntilItReleasesAssignmentToRejoin() {
+        mockStableMember();
+
+        time.sleep(DEFAULT_HEARTBEAT_INTERVAL_MS);
+        NetworkClientDelegate.PollResult result = 
heartbeatRequestManager.poll(time.milliseconds());
+        assertEquals(1, result.unsentRequests.size());
+
+        // Receive HB response fencing member
+        when(subscriptions.hasAutoAssignedPartitions()).thenReturn(true);
+        doNothing().when(membershipManager).transitionToFenced();
+        ClientResponse response = 
createHeartbeatResponse(result.unsentRequests.get(0), 
Errors.FENCED_MEMBER_EPOCH);
+        result.unsentRequests.get(0).handler().onComplete(response);
+
+        verify(membershipManager).transitionToFenced();
+        verify(heartbeatRequestState).onFailedAttempt(anyLong());
+        verify(heartbeatRequestState).reset();
+
+        when(membershipManager.state()).thenReturn(MemberState.FENCED);
+        result = heartbeatRequestManager.poll(time.milliseconds());
+        assertEquals(0, result.unsentRequests.size(), "Member should not send 
heartbeats while FENCED");
+
+        when(membershipManager.state()).thenReturn(MemberState.JOINING);
+        result = heartbeatRequestManager.poll(time.milliseconds());
+        assertEquals(1, result.unsentRequests.size(), "Fenced member should 
resume heartbeat after transitioning to JOINING");
+    }
+
     private void assertHeartbeat(HeartbeatRequestManager hrm, int nextPollMs) {
         NetworkClientDelegate.PollResult pollResult = 
hrm.poll(time.milliseconds());
         assertEquals(1, pollResult.unsentRequests.size());
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 f5c65d58bfc..50f28bb5233 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
@@ -1705,6 +1705,8 @@ public class MembershipManagerImplTest {
         assertEquals(0, listener.assignedCount());
         assertEquals(0, listener.lostCount());
 
+        assertTrue(membershipManager.shouldSkipHeartbeat(), "Member should not 
send heartbeat while fenced");
+
         // Step 3: invoke the callback
         performCallback(
                 membershipManager,