This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield 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 01f50af8b8d KAFKA-20253: Trigger rejoin on heartbeat thread
AuthenticationException (#22073)
01f50af8b8d is described below
commit 01f50af8b8de007d2474e7035e27cd331b9132bb
Author: Mingi Cho <[email protected]>
AuthorDate: Mon Jul 27 21:25:09 2026 +0900
KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException
(#22073)
[KAFKA-20253](https://issues.apache.org/jira/browse/KAFKA-20253)
Reproduced the 100% CPU busy loop using @mstruk's
[reproducer](https://github.com/mstruk/kafka-consumer-reproducer) and
tracked it down to `AbstractCoordinator.HeartbeatThread`. When an
`AuthenticationException` hits the outer catch (e.g. after a transient
OAuth server outage), it only calls `setFailureCause(e)` and exits — but
nothing triggers a rejoin. So `state` stays `STABLE`, `rejoinNeeded`
stays `false`, `heartbeatThread` gets cleared to `null`, and
`timeToNextHeartbeat()` returns the stale`heartbeatTimer.remainingMs()
== 0`. The main thread then spins `NetworkClient.poll(0)` via
`selectNow()` indefinitely.
Fixed by adding `requestRejoin()` to the auth catch block — the same
pattern already used for `REBALANCE_IN_PROGRESS` / `UNKNOWN_MEMBER_ID`
in `HeartbeatResponseHandler`. On the next `poll()` this kicks off
`ensureActiveGroup()` → `startHeartbeatThreadIfNeeded()` →
`heartbeat.resetTimeouts()`, and the consumer recovers on its own.
`GroupAuthorizationException` is left as-is since rejoining on an ACL
denial would just loop.
Unit test and validation with the reproducer:
| Scenario | CPU | Generation | | --- | --- | --- | | Before fix |
~101% (sustained) | Stuck at 5 | | After fix | ~0.3% | 5 → 15 (rejoin)
|
Thanks to @mstruk for the detailed report and the reproducer.
This change targets the Classic consumer; the related AsyncKafkaConsumer
issue is handled separately in #21714.
Reviewers: Evan Zhou <[email protected]>, Andrew Schofield
<[email protected]>
---
.../consumer/internals/AbstractCoordinator.java | 1 +
.../internals/AbstractCoordinatorTest.java | 61 ++++++++++++++++++++++
2 files changed, 62 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 6b104d37c22..b38883e5efb 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
@@ -1571,6 +1571,7 @@ public abstract class AbstractCoordinator implements
Closeable {
} 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 bf9527a4ade..2b26c9207cd 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,67 @@ public class AbstractCoordinatorTest {
}
}
+ @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");
+ // 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))
+ return false;
+ if (authErrorThrown.compareAndSet(false, true))
+ throw authError;
+ return true;
+ }, heartbeatResponse(Errors.NONE));
+
+ 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()");
+
+ 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
public void testPollHeartbeatAwakesHeartbeatThread() throws Exception {
final int longRetryBackoffMs = 10000;
