Similarly, we may have to unblock the application thread to send a {@link AsyncPollEvent} to make sure * our poll timer will not expire while we are polling. * - *
In the event that heartbeats are currently being skipped, this still returns the next heartbeat - * delay rather than {@code Long.MAX_VALUE} so that the application thread remains responsive. + *
When the member is {@link MemberState#UNSUBSCRIBED} (for example, with manual assignment), + * this returns {@code Long.MAX_VALUE} to indicate there is no next heartbeat to wait for, + * allowing the application thread to block for the full user-specified poll timeout rather than + * spinning in a busy loop. */ @Override public long maximumTimeToWait(long currentTimeMs) { pollTimer.update(currentTimeMs); + if (membershipManager().state() == MemberState.UNSUBSCRIBED) { + return Long.MAX_VALUE; + } if (pollTimer.isExpired() || (membershipManager().shouldHeartbeatNow() && !heartbeatRequestState.requestInFlight())) { return 0L; } diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java index 3ef1d712c9667..723fb9b778ce4 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManagerTest.java @@ -314,6 +314,32 @@ public void testSkippingHeartbeat(final boolean shouldSkipHeartbeat) { } } + /** + * When the consumer uses manual partition assignment (assign()) instead of subscribe(), the + * member stays in UNSUBSCRIBED state indefinitely. Because heartbeats are skipped in that + * state and heartbeatIntervalMs initialises to 0, maximumTimeToWait used to return 0, causing + * a busy-loop in pollForFetches. Verify that maximumTimeToWait returns Long.MAX_VALUE whenever + * shouldSkipHeartbeat() is true so the application thread can block for the full poll timeout. + */ + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMaximumTimeToWaitWhenHeartbeatShouldBeSkipped(final boolean shouldSkipHeartbeat) { + // Start with zero heartbeat interval (simulates the initial state before any HB response) + createHeartbeatRequestStateWithZeroHeartbeatInterval(); + when(membershipManager.shouldSkipHeartbeat()).thenReturn(shouldSkipHeartbeat); + + long result = heartbeatRequestManager.maximumTimeToWait(time.milliseconds()); + + if (shouldSkipHeartbeat) { + assertEquals(Long.MAX_VALUE, result, + "maximumTimeToWait should return Long.MAX_VALUE when heartbeats are skipped " + + "(e.g., UNSUBSCRIBED state with manual assignment) to prevent a busy loop"); + } else { + assertEquals(0, result, + "maximumTimeToWait should return 0 when heartbeat interval timer has already expired"); + } + } + @Test public void testTimerNotDue() { time.sleep(100); // time elapsed < heartbeatInterval, no heartbeat should be sent diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index d5e9014905222..7288a9465ae10 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -1669,7 +1669,11 @@ class AuthorizerIntegrationTest extends AbstractAuthorizerIntegrationTest { addAndVerifyAcls(acls, resource) waitUntilTrue(() => { - consumer.poll(Duration.ofMillis(50L)) + try { + consumer.poll(Duration.ofMillis(50L)) + } catch { + case _: TopicAuthorizationException => + } brokers.forall { broker => OptionConverters.toScala(broker.metadataCache.getLeaderAndIsr(newTopic, 0)) match { case Some(partitionState) => FetchRequest.isValidBrokerId(partitionState.leader)