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