Copilot commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3282207403
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -349,12 +358,24 @@ public MemberState state() {
/**
* @return True if the member is preparing to leave the group (waiting for
callbacks), or
* leaving (sending last heartbeat). This is used to skip proactively
leaving the group when
- * the poll timer expires.
+ * the poll timer expires. Returns {@code false} for {@link
CloseOptions.GroupMembershipOperation#REMAIN_IN_GROUP}
+ * so that {@link StreamsGroupHeartbeatRequestManager} skips the leave
heartbeat and the broker removes
+ * the member via session timeout instead.
*/
public boolean isLeavingGroup() {
+ if (CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP ==
leaveGroupOperation) {
+ return false;
+ }
return state == MemberState.PREPARE_LEAVING || state ==
MemberState.LEAVING;
}
Review Comment:
The new isLeavingGroup() implementation returns false for REMAIN_IN_GROUP
regardless of the member state. StreamsGroupHeartbeatRequestManager uses
`!membershipManager.isLeavingGroup()` to decide whether to treat
max.poll.interval expiration as a poll-timeout and proactively send a leave
heartbeat; with this change, a member closing with REMAIN_IN_GROUP can still
hit the poll-timeout path and send a leave heartbeat (or run poll-timeout
handling) once the poll timer expires. Consider keeping isLeavingGroup() purely
state-based (PREPARE_LEAVING/LEAVING) and handling the “skip leave heartbeat”
decision in the request manager (similar to
ConsumerMembershipManager/ConsumerHeartbeatRequestManager), or otherwise ensure
poll-timeout handling is bypassed during close with REMAIN_IN_GROUP.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -384,6 +385,10 @@ public NetworkClientDelegate.PollResult poll(long
currentTimeMs) {
heartbeatState.reset();
return new
NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(),
Collections.singletonList(leaveHeartbeat));
}
+ if (membershipManager.state() == MemberState.LEAVING &&
REMAIN_IN_GROUP == membershipManager.leaveGroupOperation()) {
+ membershipManager.onHeartbeatRequestSkipped();
+ return EMPTY;
+ }
Review Comment:
This REMAIN_IN_GROUP skip is checked only after the poll-timer-expired
branch above. If the poll timer expires while closing, the code can already
have generated and returned a leave heartbeat from the poll-timeout path before
reaching this check, defeating REMAIN_IN_GROUP semantics. Consider moving the
REMAIN_IN_GROUP handling ahead of the poll-timeout branch and/or making the
poll-timeout branch consult the requested membership operation before building
a leave request.
##########
streams/src/main/java/org/apache/kafka/streams/CloseOptions.java:
##########
@@ -25,22 +25,38 @@ public class CloseOptions {
* Enum to specify the group membership operation upon closing the Kafka
Streams application.
*
* <ul>
- * <li><b>{@code LEAVE_GROUP}</b>: means the consumer leave the
group.</li>
- * <li><b>{@code REMAIN_IN_GROUP}</b>: means the consumer will not leave
the group explicitly.
- * Note that this option is ignored when using the streams group
protocol
- * ({@code group.protocol=streams}); in that case, the consumer will
always leave the group.</li>
+ * <li><b>{@code DEFAULT}</b>: Applies the default behavior based on the
active protocol and membership type:
+ * <ul>
+ * <li>For the <b>classic protocol</b>: The consumer will remain in
the group.</li>
+ * <li>For the <b>streams protocol</b> ({@code
group.protocol=streams}):
+ * <ul>
+ * <li><b>Dynamic members</b>: The consumer will leave the group
(consistent with the
+ * protocol's design for dynamic members).</li>
+ * <li><b>Static members</b>: The consumer will remain in the
group until session timeout
+ * (consistent with static membership semantics).</li>
+ * </ul>
+ * </li>
+ * </ul>
+ * </li>
+ * <li><b>{@code LEAVE_GROUP}</b>: The consumer will explicitly leave
the group, regardless of the
+ * active protocol.</li>
+ * <li><b>{@code REMAIN_IN_GROUP}</b>: The consumer will remain in the
group, regardless of the
+ * active protocol. Under the streams protocol, this enables a
faster rebalance upon restart
+ * for static members.</li>
Review Comment:
The DEFAULT javadoc describes behavior for “static members” under the
streams protocol, but Streams protocol explicitly rejects static membership
(StreamsConfig.verifyStreamsProtocolCompatibility throws if group.instance.id
is set when group.protocol=streams; see StreamsConfig.java:1567-1571). This
documentation should be corrected to avoid implying static membership is
supported for 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]
