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