From 6328635602e57a4c7b313d6eb3429d0178ecf4b5 Mon Sep 17 00:00:00 2001 From: Mingi Cho <81455273+ChoMinGi@users.noreply.github.com> Date: Wed, 15 Apr 2026 17:36:16 +0900 Subject: [PATCH 1/2] KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException The AuthenticationException catch in HeartbeatThread#run only calls setFailureCause(e) and exits, so the member stays STABLE, rejoinNeeded stays false, and heartbeatThread is cleared to null. The main thread's timeToNextHeartbeat() then returns 0 from the stale heartbeatTimer, causing NetworkClient.poll(0) to spin via selectNow() indefinitely. Extend the existing "error -> requestRejoin()" pattern already used in HeartbeatResponseHandler to the auth catch, so the next poll() runs ensureActiveGroup() -> startHeartbeatThreadIfNeeded() -> heartbeat.resetTimeouts(), restoring consumer state. Verified with a new unit test and end-to-end via mstruk/kafka-consumer-reproducer (101% -> ~0.3% CPU, fresh generation). --- .../internals/AbstractCoordinator.java | 1 + .../internals/AbstractCoordinatorTest.java | 37 +++++++++++++++++++ 2 files changed, 38 insertions(+) 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 8acf9f27eb6e9..2f0a0b9df4d50 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 @@ -1575,6 +1575,7 @@ public void onFailure(RuntimeException e) { } catch (AuthenticationException e) { log.error("An authentication error occurred in the heartbeat thread", e); setFailureCause(e); + requestRejoin("authentication error in heartbeat thread"); } catch (GroupAuthorizationException e) { log.error("A group authorization error occurred in the heartbeat thread", e); setFailureCause(e); 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 18563b1ff9b30..24547c331f8ce 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 @@ -1288,6 +1288,43 @@ public void testUncaughtExceptionInHeartbeatThread() throws Exception { } } + @Test + public void testAuthenticationErrorInHeartbeatThreadTriggersRejoin() throws Exception { + setupCoordinator(); + + mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE)); + mockClient.prepareResponse(joinGroupFollowerResponse(1, memberId, leaderId, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + final AuthenticationException authError = new AuthenticationException("test auth failure"); + + mockClient.prepareResponse(body -> { + if (body instanceof HeartbeatRequest) + throw authError; + return false; + }, heartbeatResponse(Errors.UNKNOWN_SERVER_ERROR)); + + coordinator.ensureActiveGroup(); + assertFalse(coordinator.rejoinNeededOrPending(), + "Sanity: group should be stable before the auth error"); + + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + TestUtils.waitForCondition(() -> { + try { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return false; + } catch (AuthenticationException e) { + assertSame(authError, e); + return true; + } + }, 3000, "HeartbeatThread did not propagate the authentication error in time"); + + assertTrue(coordinator.rejoinNeededOrPending(), + "Expected the heartbeat thread to request a rejoin after an AuthenticationException " + + "so the next poll() restarts the heartbeat machinery via ensureActiveGroup()"); + } + @Test public void testPollHeartbeatAwakesHeartbeatThread() throws Exception { final int longRetryBackoffMs = 10000; From bdc9755f7d768667620b975740d966d015244d07 Mon Sep 17 00:00:00 2001 From: Mingi Cho <81455273+ChoMinGi@users.noreply.github.com> Date: Fri, 17 Jul 2026 17:38:28 +0900 Subject: [PATCH 2/2] KAFKA-20253: Extend heartbeat auth failure test to verify full recovery cycle - Verify rejoin completes after authentication recovers - Verify a post-recovery heartbeat succeeds without another rejoin - Ensure the prepared heartbeat response is consumed --- .../internals/AbstractCoordinatorTest.java | 30 +++++++++++++++++-- 1 file changed, 27 insertions(+), 3 deletions(-) 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 24547c331f8ce..b97a73a7878cc 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 @@ -1297,12 +1297,16 @@ public void testAuthenticationErrorInHeartbeatThreadTriggersRejoin() throws Exce mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); final AuthenticationException authError = new AuthenticationException("test auth failure"); + // A matcher exception leaves the response queued, so fail only the first heartbeat. + final AtomicBoolean authErrorThrown = new AtomicBoolean(false); mockClient.prepareResponse(body -> { - if (body instanceof HeartbeatRequest) + if (!(body instanceof HeartbeatRequest)) + return false; + if (authErrorThrown.compareAndSet(false, true)) throw authError; - return false; - }, heartbeatResponse(Errors.UNKNOWN_SERVER_ERROR)); + return true; + }, heartbeatResponse(Errors.NONE)); coordinator.ensureActiveGroup(); assertFalse(coordinator.rejoinNeededOrPending(), @@ -1323,6 +1327,26 @@ public void testAuthenticationErrorInHeartbeatThreadTriggersRejoin() throws Exce assertTrue(coordinator.rejoinNeededOrPending(), "Expected the heartbeat thread to request a rejoin after an AuthenticationException " + "so the next poll() restarts the heartbeat machinery via ensureActiveGroup()"); + + mockClient.prepareResponse(joinGroupFollowerResponse(2, memberId, leaderId, Errors.NONE)); + mockClient.prepareResponse(syncGroupResponse(Errors.NONE)); + + coordinator.ensureActiveGroup(); + + assertFalse(coordinator.rejoinNeededOrPending(), "Coordinator should have rejoined the group"); + assertEquals(2, coordinator.generation().generationId); + + mockClient.prepareResponse(body -> body instanceof HeartbeatRequest, heartbeatResponse(Errors.NONE)); + mockTime.sleep(HEARTBEAT_INTERVAL_MS); + + // Ensure the response was consumed instead of passing before a heartbeat was sent. + TestUtils.waitForCondition(() -> { + coordinator.pollHeartbeat(mockTime.milliseconds()); + return !mockClient.hasPendingResponses() && !coordinator.heartbeat().hasInflight(); + }, 3000, "Heartbeat was not sent and completed after recovering from the authentication error"); + + assertFalse(coordinator.rejoinNeededOrPending(), + "A successful post-recovery heartbeat should not trigger another rejoin"); } @Test