When a heartbeat gets FENCED_INSTANCE_ID, the heartbeat thread sets
its failure cause. Its loop then disables the thread and waits on the
coordinator monitor. The thread does not exit.

The next AbstractCoordinator.pollHeartbeat call sees the failure, sets
heartbeatThread to null, and throws the cause. It does not close the
thread. A later close() calls closeHeartbeatThread(), which finds a
null reference and returns. So the thread
kafka-coordinator-heartbeat-thread | <group> waits until the JVM exits.
An application that recreates consumers after this error collects one
such thread per consumer.

The other fatal heartbeat errors (authentication, group authorization,
interrupt, unexpected errors) leave the run loop, so those threads end.

This change makes pollHeartbeat close the failed thread before it
drops the reference. pollHeartbeat already holds the coordinator
monitor, and HeartbeatThread.close() sets the closed flag and notifies
that monitor. The thread wakes up, sees that it is closed, and returns.
pollHeartbeat does not join the thread, so it does not block.

Testing: testHeartbeatRequestWithFencedInstanceIdException now closes
the coordinator after the fenced error and waits up to 2 seconds for the
heartbeat thread to stop. Without the fix, this test fails because the
thread is still alive. With the fix, it passes. AbstractCoordinatorTest,
EagerConsumerCoordinatorTest, CooperativeConsumerCoordinatorTest, and
KafkaConsumerTest pass.

I ran the tests at trunk commit f568b7b. There, AbstractCoordinator.java
and BaseHeartbeatThread.java are the same as on current trunk.
AbstractCoordinatorTest.java differs only by the KAFKA-20858 rename in
its setup (SubscriptionState to ConsumerSubscriptionState).

Co-Authored-By: Claude Opus 5.5 noreply@anthropic.com

🤖 Generated with Claude Code