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..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
@@ -1288,6 +1288,67 @@ 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");
+        // 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;