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