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).

Author: ChoMinGi

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java

@@ -1571,6 +1571,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);

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;