lucasbru commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3302764173
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -458,20 +465,50 @@ public void resetPollTimer(final long pollMs) {
/**
* A heartbeat should be sent without waiting for the heartbeat interval
to expire if:
- * - the member is leaving the group
+ * - the member should send a leave heartbeat (see {@link
#shouldSendLeaveHeartbeat()})
* or
* - the member is joining the group or acknowledging the assignment and
for both cases there is no heartbeat request
* in flight.
*
* @return true if a heartbeat should be sent before the interval expires,
false otherwise
*/
private boolean shouldHeartbeatBeforeIntervalExpires() {
- return membershipManager.state() == MemberState.LEAVING
- ||
- (membershipManager.state() == MemberState.JOINING ||
membershipManager.state() == MemberState.ACKNOWLEDGING)
+ return shouldSendLeaveHeartbeat()
+ || (membershipManager.state() == MemberState.JOINING ||
membershipManager.state() == MemberState.ACKNOWLEDGING)
&& !heartbeatRequestState.requestInFlight();
}
+ /**
+ * Returns whether a leave group heartbeat should be sent. The leave
heartbeat is skipped when:
+ * <ul>
+ * <li>The operation is {@link
org.apache.kafka.clients.consumer.CloseOptions.GroupMembershipOperation#REMAIN_IN_GROUP},
or</li>
+ * <li>The operation is {@link
org.apache.kafka.clients.consumer.CloseOptions.GroupMembershipOperation#DEFAULT}
and the member is static
+ * (has a group instance ID), since the broker will remove the
static member via session timeout.</li>
+ * </ul>
+ *
+ * @return true if a leave heartbeat should be sent, false otherwise
+ */
+ private boolean shouldSendLeaveHeartbeat() {
+ if (shouldSkipLeaveHeartbeat()) {
+ logger.debug("Member {} skipping leave heartbeat (operation={},
static={}).",
+ membershipManager.memberId(),
+ membershipManager.leaveGroupOperation(),
+ membershipManager.groupInstanceId().isPresent());
+ return false;
+ }
+ return membershipManager.state() == MemberState.LEAVING;
+ }
+
+ /**
+ * Returns true if the leave heartbeat should be skipped: either the
operation is REMAIN_IN_GROUP,
+ * or the operation is DEFAULT and the member is static (has a group
instance ID).
+ */
+ private boolean shouldSkipLeaveHeartbeat() {
Review Comment:
Wondering if we should keep this closer to the consumer pattern. In
ConsumerHeartbeatRequestManager.shouldSendLeaveHeartbeatNow we only skip the
leave HB for dynamic + REMAIN_IN_GROUP — for static + REMAIN/DEFAULT we still
send a leave heartbeat, but with epoch -2 (LEAVE_GROUP_STATIC_MEMBER_EPOCH) so
the broker knows "static member temporarily gone, hold the assignment". Here we
skip the HB entirely for those cases and the broker only finds out at session
timeout. The -2 constant already exists on StreamsGroupHeartbeatRequest but the
client never sends it.
To mirror the consumer side roughly:
- shouldSkipLeaveHeartbeat: skip only for dynamic + REMAIN_IN_GROUP
- isLeavingGroup: combine state + operation + static (like
ConsumerMembershipManager.isLeavingGroup), instead of being purely state-based
- add a leaveGroupEpoch() that returns -2 for static + (REMAIN/DEFAULT),
else -1
- use it where we currently hardcode LEAVE_GROUP_MEMBER_EPOCH
- the `intentional` boolean on onHeartbeatRequestSkipped would also go away
(only intentional skip becomes dynamic + REMAIN)
Broker-side handling of -2 in streams-group HB is a separate task
(GroupMetadataManager.streamsGroupHeartbeat currently throws "Static members
are not supported yet."), but I'd rather have the client wired the same way as
the consumer so when broker support lands it just works.
--
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]
