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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
