KAFKA-20970 Another busy loop happens if auto.commit.interval.ms is less than bootstrap.resolve.timeout.ms

[m1a2st]

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.

Reviewers: Chia-Ping Tsai chia7712@gmail.com, Andrew Schofield
aschofield@confluent.io

[chia7712]

@m1a2st Would you mind enabling DNS resolution in this patch? It can help us catch unexpected latency or resource issues

[AndrewJSchofield]

Might StreamsGroupHeartbeatRequestManager needs a similar check for coordinatorRequestManager.coordinator().isEmpty()? It seems like an outlier.

[AndrewJSchofield]

Thanks for the PR. Looks good to me. Just one suggestion for the test.

[AndrewJSchofield]

I wonder whether when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty()); might be preferable here. Makes it clear what is being asserted.

[m1a2st]

Addressed it.

[m1a2st]

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.