You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/08/21 13:57:00 UTC
[PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
m1a2st opened a new pull request, #23227:
URL: https://github.com/apache/kafka/pull/23227
`CommitRequestManager#maximumTimeToWait()` returned the raw remaining time of the
auto-commit timer, which becomes `0` once the interval elapses. However, `poll()`
cannot actually send an auto-commit until the coordinator is known; it returns
`EMPTY` while `coordinatorRequestManager.coordinator()` is empty. When bootstrap
DNS resolution is still in progress (`bootstrap.resolve.timeout.ms > 0`) and
`auto.commit.interval.ms` is shorter than the resolution window, the auto-commit
timer can expire before the coordinator is discovered. `remainingMs()` then
continues to return `0` indefinitely, since the timer is not reset until the
application thread successfully sends an auto-commit.
This `0` is propagated to `ConsumerNetworkThread#cachedMaximumTimeToWait`, which
bounds the `pollTimeout` in `AsyncKafkaConsumer#pollForFetches()` to `0`, causing
the application and network threads to busy-loop for as long as the coordinator
remains unknown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3834873224
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +731,18 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+
Review Comment:
Addressed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3913714539
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result = commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration = BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit timer expires several
+ // times while the coordinator is still (and will remain, since DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config, subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata, "test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000, 30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new CommitRequestManager(
+ time, logContext, subscriptionState, config, realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID, Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false, mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs + 3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
Review Comment:
Yes, MockSelector addresses that, its `poll(timeout)` calls `time.sleep(timeout)`
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result = commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration = BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit timer expires several
+ // times while the coordinator is still (and will remain, since DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config, subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata, "test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000, 30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new CommitRequestManager(
+ time, logContext, subscriptionState, config, realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID, Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false, mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs + 3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
+ // Drives the real NetworkClient's ensureBootstrapped()/async DNS resolution forward;
+ // the coordinator never becomes known since there is no real broker to respond.
+ networkClientDelegate.poll(50, time.milliseconds());
+
+ Optional<Exception> metadataError = networkClientDelegate.getAndClearMetadataError();
+ if (metadataError.isPresent()) {
+ assertInstanceOf(BootstrapResolutionException.class, metadataError.get());
+ sawBootstrapException = true;
+ break;
+ }
+
+ long waitMs = realCommitRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(waitMs > 0, "maximumTimeToWait must be > 0 while real bootstrap DNS resolution is pending; got " + waitMs);
Review Comment:
Sure, addressed it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #23227:
URL: https://github.com/apache/kafka/pull/23227#issuecomment-5377430870
> Might StreamsGroupHeartbeatRequestManager needs a similar check for coordinatorRequestManager.coordinator().isEmpty()? It seems like an outlier.
Good catch. I'll address this in this patch as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3904470294
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -530,9 +530,19 @@ public StreamsMembershipManager membershipManager() {
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (pollTimer.isExpired() ||
- membershipManager.shouldNotWaitForHeartbeatInterval() && !heartbeatRequestState.requestInFlight()) {
-
+ if (pollTimer.isExpired()) {
+ return 0L;
+ }
+ // A heartbeat is only sent when the coordinator is known; poll() returns EMPTY otherwise
+ // (see the guard at the top of poll()). If the coordinator is unavailable (for example,
+ // while bootstrap DNS resolution is still in progress), the
+ // shouldNotWaitForHeartbeatInterval() check would return 0 whenever the member wants to
+ // (re)join. Because no heartbeat can be sent until the coordinator is discovered, the
+ // condition remains true and both the application and network threads end up busy-spinning.
+ if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
Review Comment:
If the bootstrap DNS resolution is not finished, `coordinatorRequestManager.coordinator().isEmpty()` will be true. The initial value of `heartbeatIntervalMs` is zero, so the poll timeout will be adjusted to zero, causing a busy loop.
I think this is a broader issue since `AbstractHeartbeatRequestManager` is impacted too, which means the streams consumer, async consumer, and share consumer all have the same issue.
Plus, the straightforward way is to return `retryBackoffMs` instead of `heartbeatIntervalMs`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #23227:
URL: https://github.com/apache/kafka/pull/23227#issuecomment-5499992492
opened https://issues.apache.org/jira/browse/KAFKA-21014 for the flaky test
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3831435557
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +731,18 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+
Review Comment:
I wonder whether `when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
` might be preferable here. Makes it clear what is being asserted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3918767430
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -221,7 +221,19 @@ public void signalClose() {
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
- return autoCommitState.map(ac -> ac.remainingMs(currentTimeMs)).orElse(Long.MAX_VALUE);
+ if (autoCommitState.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ AutoCommitState autoCommit = autoCommitState.get();
+ // An auto-commit is only sent when the coordinator is known; poll() returns EMPTY otherwise.
+ // If the coordinator is unavailable (e.g. bootstrap DNS resolution is still in progress),
+ // falling through to the timer-based remainingMs() would return 0 once the auto-commit interval
+ // elapses, since the auto-commit timer remains permanently expired. This would cause both the
+ // application and network threads to busy-spin.
+ if (coordinatorRequestManager.coordinator().isEmpty()) {
+ return autoCommit.autoCommitIntervalMs();
Review Comment:
> In the fetch path we used the retryBackoff for the case of inflights (which is smaller than the commit internval by default)
Agreed. Since the same question applies to all the empty coordinator guards, could we defer this to KAFKA-21010 and settle on a consistent return value there? WDYT
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3902549110
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -221,7 +221,19 @@ public void signalClose() {
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
- return autoCommitState.map(ac -> ac.remainingMs(currentTimeMs)).orElse(Long.MAX_VALUE);
+ if (autoCommitState.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ AutoCommitState autocommit = autoCommitState.get();
Review Comment:
```suggestion
AutoCommitState autoCommit = autoCommitState.get();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3904528109
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -530,9 +530,19 @@ public StreamsMembershipManager membershipManager() {
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (pollTimer.isExpired() ||
- membershipManager.shouldNotWaitForHeartbeatInterval() && !heartbeatRequestState.requestInFlight()) {
-
+ if (pollTimer.isExpired()) {
+ return 0L;
+ }
+ // A heartbeat is only sent when the coordinator is known; poll() returns EMPTY otherwise
+ // (see the guard at the top of poll()). If the coordinator is unavailable (for example,
+ // while bootstrap DNS resolution is still in progress), the
+ // shouldNotWaitForHeartbeatInterval() check would return 0 whenever the member wants to
+ // (re)join. Because no heartbeat can be sent until the coordinator is discovered, the
+ // condition remains true and both the application and network threads end up busy-spinning.
+ if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
Review Comment:
@AndrewJSchofield @frankvicky I'd prefer to keep the current change as is, since it's consistent with the guard in AbstractHeartbeatRequestManager from KAFKA-20253. What the guard should actually return (for example retryBackoffMs) deserves its own discussion, so I opened KAFKA-21010 to address it for all three heartbeat managers together. Note that this means the streams change here still doesn't cover the pre-join case yet, but that is no worse than trunk today. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #23227:
URL: https://github.com/apache/kafka/pull/23227#issuecomment-5529789810
cherry-pick to 4.4
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3925488250
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -221,7 +221,19 @@ public void signalClose() {
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
- return autoCommitState.map(ac -> ac.remainingMs(currentTimeMs)).orElse(Long.MAX_VALUE);
+ if (autoCommitState.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ AutoCommitState autoCommit = autoCommitState.get();
+ // An auto-commit is only sent when the coordinator is known; poll() returns EMPTY otherwise.
+ // If the coordinator is unavailable (e.g. bootstrap DNS resolution is still in progress),
+ // falling through to the timer-based remainingMs() would return 0 once the auto-commit interval
+ // elapses, since the auto-commit timer remains permanently expired. This would cause both the
+ // application and network threads to busy-spin.
+ if (coordinatorRequestManager.coordinator().isEmpty()) {
+ return autoCommit.autoCommitIntervalMs();
Review Comment:
> could we defer this to [KAFKA-21010](https://issues.apache.org/jira/browse/KAFKA-21010) and settle on a consistent return value there?
Sure, sg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky merged PR #23227:
URL: https://github.com/apache/kafka/pull/23227
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3920511779
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -221,7 +221,19 @@ public void signalClose() {
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
- return autoCommitState.map(ac -> ac.remainingMs(currentTimeMs)).orElse(Long.MAX_VALUE);
+ if (autoCommitState.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ AutoCommitState autoCommit = autoCommitState.get();
+ // An auto-commit is only sent when the coordinator is known; poll() returns EMPTY otherwise.
+ // If the coordinator is unavailable (e.g. bootstrap DNS resolution is still in progress),
+ // falling through to the timer-based remainingMs() would return 0 once the auto-commit interval
+ // elapses, since the auto-commit timer remains permanently expired. This would cause both the
+ // application and network threads to busy-spin.
+ if (coordinatorRequestManager.coordinator().isEmpty()) {
+ return autoCommit.autoCommitIntervalMs();
Review Comment:
Thanks for the feedback. I agree that we should use `retryBackoffMs` instead of `autoCommitIntervalMs`, and I can file a patch to address it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3904479142
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -530,9 +530,19 @@ public StreamsMembershipManager membershipManager() {
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (pollTimer.isExpired() ||
- membershipManager.shouldNotWaitForHeartbeatInterval() && !heartbeatRequestState.requestInFlight()) {
-
+ if (pollTimer.isExpired()) {
+ return 0L;
+ }
+ // A heartbeat is only sent when the coordinator is known; poll() returns EMPTY otherwise
+ // (see the guard at the top of poll()). If the coordinator is unavailable (for example,
+ // while bootstrap DNS resolution is still in progress), the
+ // shouldNotWaitForHeartbeatInterval() check would return 0 whenever the member wants to
+ // (re)join. Because no heartbeat can be sent until the coordinator is discovered, the
+ // condition remains true and both the application and network threads end up busy-spinning.
+ if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
Review Comment:
opened https://issues.apache.org/jira/browse/KAFKA-21010
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3904747841
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -530,9 +530,19 @@ public StreamsMembershipManager membershipManager() {
@Override
public long maximumTimeToWait(long currentTimeMs) {
pollTimer.update(currentTimeMs);
- if (pollTimer.isExpired() ||
- membershipManager.shouldNotWaitForHeartbeatInterval() && !heartbeatRequestState.requestInFlight()) {
-
+ if (pollTimer.isExpired()) {
+ return 0L;
+ }
+ // A heartbeat is only sent when the coordinator is known; poll() returns EMPTY otherwise
+ // (see the guard at the top of poll()). If the coordinator is unavailable (for example,
+ // while bootstrap DNS resolution is still in progress), the
+ // shouldNotWaitForHeartbeatInterval() check would return 0 whenever the member wants to
+ // (re)join. Because no heartbeat can be sent until the coordinator is discovered, the
+ // condition remains true and both the application and network threads end up busy-spinning.
+ if (coordinatorRequestManager.coordinator().isEmpty() || membershipManager.shouldSkipHeartbeat()) {
Review Comment:
Sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3916789355
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java:
##########
@@ -221,7 +221,19 @@ public void signalClose() {
*/
@Override
public long maximumTimeToWait(long currentTimeMs) {
- return autoCommitState.map(ac -> ac.remainingMs(currentTimeMs)).orElse(Long.MAX_VALUE);
+ if (autoCommitState.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ AutoCommitState autoCommit = autoCommitState.get();
+ // An auto-commit is only sent when the coordinator is known; poll() returns EMPTY otherwise.
+ // If the coordinator is unavailable (e.g. bootstrap DNS resolution is still in progress),
+ // falling through to the timer-based remainingMs() would return 0 once the auto-commit interval
+ // elapses, since the auto-commit timer remains permanently expired. This would cause both the
+ // application and network threads to busy-spin.
+ if (coordinatorRequestManager.coordinator().isEmpty()) {
+ return autoCommit.autoCommitIntervalMs();
Review Comment:
agree with waiting a bit here, just wondering if the interval is the best "delay" to use?
In the fetch path we used the `retryBackoff` for the case of inflights (which is smaller than the commit internval by default)
https://github.com/apache/kafka/blob/820533b870106cc0e0ac60e2076b8644d68bd85f/clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java#L84
Is there a reason for preferring the interval here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3913062017
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result = commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration = BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit timer expires several
+ // times while the coordinator is still (and will remain, since DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config, subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata, "test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000, 30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new CommitRequestManager(
+ time, logContext, subscriptionState, config, realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID, Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false, mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs + 3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
Review Comment:
I was mulling over how time gets advanced ... Does `MockSelector` address that? If so, that's fine.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+ when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result = commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration = BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit timer expires several
+ // times while the coordinator is still (and will remain, since DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config, subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata, "test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000, 30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new CommitRequestManager(
+ time, logContext, subscriptionState, config, realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID, Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false, mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs + 3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
+ // Drives the real NetworkClient's ensureBootstrapped()/async DNS resolution forward;
+ // the coordinator never becomes known since there is no real broker to respond.
+ networkClientDelegate.poll(50, time.milliseconds());
+
+ Optional<Exception> metadataError = networkClientDelegate.getAndClearMetadataError();
+ if (metadataError.isPresent()) {
+ assertInstanceOf(BootstrapResolutionException.class, metadataError.get());
+ sawBootstrapException = true;
+ break;
+ }
+
+ long waitMs = realCommitRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(waitMs > 0, "maximumTimeToWait must be > 0 while real bootstrap DNS resolution is pending; got " + waitMs);
Review Comment:
Should we check `waitMs` before breaking the loop
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #23227:
URL: https://github.com/apache/kafka/pull/23227#issuecomment-5371425358
Might `StreamsGroupHeartbeatRequestManager` needs a similar check for `coordinatorRequestManager.coordinator().isEmpty()`? It seems like an outlier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org