m1a2st commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3304800957
##########
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:
Thanks for the review! I've updated the flow to mirror
ConsumerHeartbeatRequestManager's pattern, The client is now wired the same way
as the consumer, so when broker-side support for epoch -2 in the streams
protocol.
--
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]
