ableegoldman commented on code in PR #17614:
URL: https://github.com/apache/kafka/pull/17614#discussion_r2025873637
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -211,6 +214,28 @@ public ConsumerMembershipManager membershipManager() {
return membershipManager;
}
+ @Override
+ protected boolean isLeavingGroup() {
+ // If the consumer has dynamic membership,
+ // we should skip the leaving heartbeat when leaveGroupOperation is
REMAIN_IN_GROUP
+ if (membershipManager.groupInstanceId().isEmpty() && REMAIN_IN_GROUP
== membershipManager.leaveGroupOperation())
+ return false;
+ return membershipManager().state() == MemberState.LEAVING;
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult pollOnClose(long currentTimeMs) {
+ // If the consumer has dynamic membership,
+ // we should skip the leaving heartbeat when leaveGroupOperation is
REMAIN_IN_GROUP
+ boolean skipHeartbeatForDynamicMemberRemainInGroup =
membershipManager.groupInstanceId().isEmpty()
+ && REMAIN_IN_GROUP == membershipManager.leaveGroupOperation();
Review Comment:
very minor nit: the double negative with "!skipX" is a bit confusing,
instead of "skipHeartbeat..." can we invert the variable to "shouldHeartbeat"?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractHeartbeatRequestManager.java:
##########
@@ -182,7 +181,7 @@ public NetworkClientDelegate.PollResult poll(long
currentTimeMs) {
}
// Case 1: The member is leaving
- boolean heartbeatNow = membershipManager().state() ==
MemberState.LEAVING ||
+ boolean heartbeatNow = isLeavingGroup() ||
Review Comment:
why add the `#isLeavingGroup` method to the AbstractHeartbeatRequestManager
instead of just calling `membershipManager#isLeavingGroup` here?
Basically I'm wondering why both the HeartbeatRequestManager and
MembershipManager classes need to have this `#isLeavingGroup` method
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
