Skip to content

KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException #22073

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ChoMinGi wants to merge 2 commits into
base: trunk
Choose a base branch
from
+62 −0
Open

KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException #22073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6328635
KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException
ChoMinGi Apr 15, 2026
bdc9755
KAFKA-20253: Extend heartbeat auth failure test to verify full recove…
ChoMinGi Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
61 changes: 61 additions & 0 deletions ...ts/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1288,6 +1288,67 @@ public void testUncaughtExceptionInHeartbeatThread() throws Exception {
}
}

@Test
public void testAuthenticationErrorInHeartbeatThreadTriggersRejoin() throws Exception {

@AndrewJSchofield AndrewJSchofield Jul 16, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This test in a way tests that the chosen fix was implemented, namely that rejoin was requested. What I'd really like to see is a test that does an authentication failure on the first heartbeat, then the rejoin is requested, and a subsequent successful heartbeat works as expected.

@ChoMinGi ChoMinGi Jul 17, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@AndrewJSchofield

Done in bdc9755 — The test now covers the full cycle.
auth failure on the first heartbeat, rejoin requested, successful JoinGroup/SyncGroup once auth recovers, then a subsequent heartbeat completing without another rejoin.

To keep the last assertion from passing vacuously, the test requires the prepared heartbeat response to be consumed with no heartbeat left in flight — the matcher only accepts a HeartbeatRequest, so the post-recovery heartbeat must actually have been sent and processed.

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;
Expand Down
Loading