frankvicky commented on code in PR #17614:
URL: https://github.com/apache/kafka/pull/17614#discussion_r2029880923
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -211,6 +214,29 @@ public ConsumerMembershipManager membershipManager() {
return membershipManager;
}
+ @Override
+ protected boolean shouldSendLeaveHeartbeatNow() {
+ // 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) {
+ // Determine if we should send a leaving heartbeat:
+ // - For static membership (when groupInstanceId is present): Always
send the leaving heartbeat
+ // - For dynamic membership: Send the leaving heartbeat only when
leaveGroupOperation is not REMAIN_IN_GROUP
+ boolean shouldHeartbeat =
membershipManager.groupInstanceId().isPresent()
Review Comment:
Yes, this could significantly reduce redundancy!
Previously, I thought it was more clear that the async consumer and share
consumer have their implementation, but it does not provide too much value now.
--
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
