You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/02/25 16:14:39 UTC
[PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
m1a2st opened a new pull request, #21579:
URL: https://github.com/apache/kafka/pull/21579
This patch is addressed the https://cwiki.apache.org/confluence/display/KAFKA/KIP-1284+Introduce+CloseOptions.DEFAULT+for+Kafka+Streams
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "chickenchickenlove (via GitHub)" <gi...@apache.org>.
chickenchickenlove commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3179925158
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -899,23 +900,43 @@ private static Map<String, SortedSet<Integer>> toTasksAssignment(final List<Stre
}
/**
- * Leaves the group when the member closes.
+ * Closes the member's participation in the group, honoring the requested {@link CloseOptions.GroupMembershipOperation}:
*
- * <p>
- * This method does the following:
- * <ol>
- * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Skips the invocation of the revocation callback or lost callback.</li>
- * <li>Clears the current and target assignment, unsubscribes from all topics and
- * transitions the member state to {@link MemberState#LEAVING}.</li>
- * </ol>
- * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
- * to send a leave group heartbeat.
- * </p>
+ * <ul>
+ * <li>{@code REMAIN_IN_GROUP}: clears local task assignments, unsubscribes, and completes
+ * immediately without sending a leave group heartbeat. The broker will remove the member
+ * via session timeout.</li>
+ * <li>{@code DEFAULT} / {@code LEAVE_GROUP}: transitions to
+ * {@link MemberState#PREPARE_LEAVING} → {@link MemberState#LEAVING} and sends the leave
+ * group heartbeat.</li>
+ * </ul>
*
- * @return future that will complete when the heartbeat to leave the group has been sent out.
+ * @param membershipOperation the requested close behavior
+ * @return future that will complete when the close operation is done
*/
- public CompletableFuture<Void> leaveGroupOnClose() {
+ public CompletableFuture<Void> leaveGroupOnClose(final CloseOptions.GroupMembershipOperation membershipOperation) {
+ if (CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP == membershipOperation) {
+ // Skip the leave group heartbeat. Clean up locally and complete immediately.
+ // The broker will remove the member from the group via session timeout.
+ log.info("Skipping leave group heartbeat for member {} with epoch {} because REMAIN_IN_GROUP was specified.",
+ memberId, memberEpoch);
+ if (isNotInGroup()) {
+ subscriptionState.unsubscribe();
+ notifyAssignmentChange(Collections.emptySet());
+ return CompletableFuture.completedFuture(null);
+ }
+ clearTaskAndPartitionAssignment();
+ subscriptionState.unsubscribe();
+ // UNSUBSCRIBED is only reachable from PREPARE_LEAVING, LEAVING, or FENCED.
+ // For any active state (JOINING/STABLE/RECONCILING/ACKNOWLEDGING) we must pass
+ // through PREPARE_LEAVING first; if already in PREPARE_LEAVING or LEAVING we
+ // can transition directly.
+ if (state != MemberState.PREPARE_LEAVING && state != MemberState.LEAVING) {
+ transitionTo(MemberState.PREPARE_LEAVING);
+ }
+ transitionTo(MemberState.UNSUBSCRIBED);
+ return CompletableFuture.completedFuture(null);
+ }
Review Comment:
Even in the `REMAIN_IN_GROUP` case, I think it would be better to follow the same normal leaving path as the existing `private CompletableFuture<Void> leaveGroup(final boolean isOnClose) { ... }`.
In the `ConsumerGroup` case, `leaveGroupOnClose()` stores the close operation in `leaveGroupOperation`, then transitions through the normal leaving path to the LEAVING state. At the heartbeat request generation stage, `ConsumerHeartbeatRequestManager#shouldSendLeaveHeartbeatNow()` looks at this operation and decides whether to send a leave heartbeat.
In particular, it is intentionally designed to skip the leave heartbeat only for the dynamic member + `REMAIN_IN_GROUP` case.
For `StreamsGroup` as well, rather than handling `REMAIN_IN_GROUP` here with an early return, I think it would be more aligned with the `ConsumerGroup` flow to preserve the normal leaving path and let `StreamsGroupHeartbeatRequestManager` decide whether to send or skip the heartbeat based on the operation.
I think this structure would also be more natural when implementing client-side support for Streams static membership later.
##########
streams/src/main/java/org/apache/kafka/streams/CloseOptions.java:
##########
@@ -25,22 +25,32 @@ 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:
+ * <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}): The consumer will leave
+ * the group (consistent with the protocol's design for dynamic members).</li>
+ * </ul>
Review Comment:
Should we update this to clarify that CloseOptions.DEFAULT behaves differently for static members and dynamic members?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -899,23 +900,43 @@ private static Map<String, SortedSet<Integer>> toTasksAssignment(final List<Stre
}
/**
- * Leaves the group when the member closes.
+ * Closes the member's participation in the group, honoring the requested {@link CloseOptions.GroupMembershipOperation}:
*
- * <p>
- * This method does the following:
- * <ol>
- * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Skips the invocation of the revocation callback or lost callback.</li>
- * <li>Clears the current and target assignment, unsubscribes from all topics and
- * transitions the member state to {@link MemberState#LEAVING}.</li>
- * </ol>
- * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
- * to send a leave group heartbeat.
- * </p>
+ * <ul>
+ * <li>{@code REMAIN_IN_GROUP}: clears local task assignments, unsubscribes, and completes
+ * immediately without sending a leave group heartbeat. The broker will remove the member
+ * via session timeout.</li>
+ * <li>{@code DEFAULT} / {@code LEAVE_GROUP}: transitions to
+ * {@link MemberState#PREPARE_LEAVING} → {@link MemberState#LEAVING} and sends the leave
+ * group heartbeat.</li>
+ * </ul>
*
- * @return future that will complete when the heartbeat to leave the group has been sent out.
+ * @param membershipOperation the requested close behavior
+ * @return future that will complete when the close operation is done
*/
- public CompletableFuture<Void> leaveGroupOnClose() {
+ public CompletableFuture<Void> leaveGroupOnClose(final CloseOptions.GroupMembershipOperation membershipOperation) {
+ if (CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP == membershipOperation) {
+ // Skip the leave group heartbeat. Clean up locally and complete immediately.
+ // The broker will remove the member from the group via session timeout.
+ log.info("Skipping leave group heartbeat for member {} with epoch {} because REMAIN_IN_GROUP was specified.",
Review Comment:
How about logging this after `isNotIngroup()`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -899,23 +900,43 @@ private static Map<String, SortedSet<Integer>> toTasksAssignment(final List<Stre
}
/**
- * Leaves the group when the member closes.
+ * Closes the member's participation in the group, honoring the requested {@link CloseOptions.GroupMembershipOperation}:
*
- * <p>
- * This method does the following:
- * <ol>
- * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Skips the invocation of the revocation callback or lost callback.</li>
- * <li>Clears the current and target assignment, unsubscribes from all topics and
- * transitions the member state to {@link MemberState#LEAVING}.</li>
- * </ol>
- * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
- * to send a leave group heartbeat.
- * </p>
+ * <ul>
+ * <li>{@code REMAIN_IN_GROUP}: clears local task assignments, unsubscribes, and completes
+ * immediately without sending a leave group heartbeat. The broker will remove the member
+ * via session timeout.</li>
+ * <li>{@code DEFAULT} / {@code LEAVE_GROUP}: transitions to
+ * {@link MemberState#PREPARE_LEAVING} → {@link MemberState#LEAVING} and sends the leave
+ * group heartbeat.</li>
+ * </ul>
*
- * @return future that will complete when the heartbeat to leave the group has been sent out.
+ * @param membershipOperation the requested close behavior
+ * @return future that will complete when the close operation is done
*/
- public CompletableFuture<Void> leaveGroupOnClose() {
+ public CompletableFuture<Void> leaveGroupOnClose(final CloseOptions.GroupMembershipOperation membershipOperation) {
Review Comment:
`ConsumerGroup` stores the `leaveGroupOperation` here and follows the normal leaving path. Then, `AbstractHeartbeatRequestManager#poll()` decides whether to send or skip the Heartbeat request.
Similarly, I think `StreamsGroup` should also add a `leaveGroupOperation` field here. That way, `StreamsGroup` can follow the normal leaving path as well, and `StreamsGroupHeartbeatRequestManager can` decide whether to send or skip the Heartbeat request.
What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3195678535
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -899,23 +900,43 @@ private static Map<String, SortedSet<Integer>> toTasksAssignment(final List<Stre
}
/**
- * Leaves the group when the member closes.
+ * Closes the member's participation in the group, honoring the requested {@link CloseOptions.GroupMembershipOperation}:
*
- * <p>
- * This method does the following:
- * <ol>
- * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Skips the invocation of the revocation callback or lost callback.</li>
- * <li>Clears the current and target assignment, unsubscribes from all topics and
- * transitions the member state to {@link MemberState#LEAVING}.</li>
- * </ol>
- * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
- * to send a leave group heartbeat.
- * </p>
+ * <ul>
+ * <li>{@code REMAIN_IN_GROUP}: clears local task assignments, unsubscribes, and completes
+ * immediately without sending a leave group heartbeat. The broker will remove the member
+ * via session timeout.</li>
+ * <li>{@code DEFAULT} / {@code LEAVE_GROUP}: transitions to
+ * {@link MemberState#PREPARE_LEAVING} → {@link MemberState#LEAVING} and sends the leave
+ * group heartbeat.</li>
+ * </ul>
*
- * @return future that will complete when the heartbeat to leave the group has been sent out.
+ * @param membershipOperation the requested close behavior
+ * @return future that will complete when the close operation is done
*/
- public CompletableFuture<Void> leaveGroupOnClose() {
+ public CompletableFuture<Void> leaveGroupOnClose(final CloseOptions.GroupMembershipOperation membershipOperation) {
+ if (CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP == membershipOperation) {
+ // Skip the leave group heartbeat. Clean up locally and complete immediately.
+ // The broker will remove the member from the group via session timeout.
+ log.info("Skipping leave group heartbeat for member {} with epoch {} because REMAIN_IN_GROUP was specified.",
+ memberId, memberEpoch);
+ if (isNotInGroup()) {
+ subscriptionState.unsubscribe();
+ notifyAssignmentChange(Collections.emptySet());
+ return CompletableFuture.completedFuture(null);
+ }
+ clearTaskAndPartitionAssignment();
+ subscriptionState.unsubscribe();
+ // UNSUBSCRIBED is only reachable from PREPARE_LEAVING, LEAVING, or FENCED.
+ // For any active state (JOINING/STABLE/RECONCILING/ACKNOWLEDGING) we must pass
+ // through PREPARE_LEAVING first; if already in PREPARE_LEAVING or LEAVING we
+ // can transition directly.
+ if (state != MemberState.PREPARE_LEAVING && state != MemberState.LEAVING) {
+ transitionTo(MemberState.PREPARE_LEAVING);
+ }
+ transitionTo(MemberState.UNSUBSCRIBED);
+ return CompletableFuture.completedFuture(null);
+ }
Review Comment:
Agreed. Moved the decision to `StreamsGroupHeartbeatRequestManager:` added `shouldSendLeaveHeartbeat()` that checks `leaveGroupOperation()`, and updated `poll()` to call `onHeartbeatRequestSkipped()` directly for `REMAIN_IN_GROUP` in `LEAVING` state. `isLeavingGroup()` also returns false for `REMAIN_IN_GROUP` so `pollOnClose()` won't force a close heartbeat. This mirrors the ConsumerGroup pattern and should also make it easier to layer in static membership support later.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -899,23 +900,43 @@ private static Map<String, SortedSet<Integer>> toTasksAssignment(final List<Stre
}
/**
- * Leaves the group when the member closes.
+ * Closes the member's participation in the group, honoring the requested {@link CloseOptions.GroupMembershipOperation}:
*
- * <p>
- * This method does the following:
- * <ol>
- * <li>Transitions member state to {@link MemberState#PREPARE_LEAVING}.</li>
- * <li>Skips the invocation of the revocation callback or lost callback.</li>
- * <li>Clears the current and target assignment, unsubscribes from all topics and
- * transitions the member state to {@link MemberState#LEAVING}.</li>
- * </ol>
- * States {@link MemberState#PREPARE_LEAVING} and {@link MemberState#LEAVING} cause the heartbeat request manager
- * to send a leave group heartbeat.
- * </p>
+ * <ul>
+ * <li>{@code REMAIN_IN_GROUP}: clears local task assignments, unsubscribes, and completes
+ * immediately without sending a leave group heartbeat. The broker will remove the member
+ * via session timeout.</li>
+ * <li>{@code DEFAULT} / {@code LEAVE_GROUP}: transitions to
+ * {@link MemberState#PREPARE_LEAVING} → {@link MemberState#LEAVING} and sends the leave
+ * group heartbeat.</li>
+ * </ul>
*
- * @return future that will complete when the heartbeat to leave the group has been sent out.
+ * @param membershipOperation the requested close behavior
+ * @return future that will complete when the close operation is done
*/
- public CompletableFuture<Void> leaveGroupOnClose() {
+ public CompletableFuture<Void> leaveGroupOnClose(final CloseOptions.GroupMembershipOperation membershipOperation) {
Review Comment:
Thanks for the suggestion! I've added a `leaveGroupOperation` field to `StreamsMembershipManager` and simplified `leaveGroupOnClose()` to just store the operation and delegate to `leaveGroup(true)`. The member now always follows the normal leaving path.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21579:
URL: https://github.com/apache/kafka/pull/21579#issuecomment-4694521039
@m1a2st -- Seems we did not update the docs for this change. Each KIP should be listed in the streams upgrade guide. Would you be interested to do a follow up PR for it?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3302710910
##########
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:
This will resolve with static membership being merged
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21579:
URL: https://github.com/apache/kafka/pull/21579#issuecomment-4525366074
Hi @lucasbru, thanks for the detailed review! I've addressed all your comments.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #21579:
URL: https://github.com/apache/kafka/pull/21579#issuecomment-4696755332
Thanks for reminder, I will open a followup patch for it
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3311230684
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -384,6 +385,13 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
heartbeatState.reset();
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(), Collections.singletonList(leaveHeartbeat));
}
+ if (membershipManager.state() == MemberState.LEAVING && shouldSkipLeaveHeartbeat()) {
+ logger.info("Dynamic member {} skipping leave heartbeat (operation=REMAIN_IN_GROUP). " +
+ "The broker will remove the member from the group via session timeout.",
+ membershipManager.memberId());
+ membershipManager.onHeartbeatRequestSkipped();
Review Comment:
Thanks! I went with guarding the WARN and reused the existing `isLeavingGroup()` helper 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams (WIP) [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #21579:
URL: https://github.com/apache/kafka/pull/21579#issuecomment-4001940471
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3309556887
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -384,6 +385,13 @@ public NetworkClientDelegate.PollResult poll(long currentTimeMs) {
heartbeatState.reset();
return new NetworkClientDelegate.PollResult(heartbeatRequestState.heartbeatIntervalMs(), Collections.singletonList(leaveHeartbeat));
}
+ if (membershipManager.state() == MemberState.LEAVING && shouldSkipLeaveHeartbeat()) {
+ logger.info("Dynamic member {} skipping leave heartbeat (operation=REMAIN_IN_GROUP). " +
+ "The broker will remove the member from the group via session timeout.",
+ membershipManager.memberId());
+ membershipManager.onHeartbeatRequestSkipped();
Review Comment:
nit: onHeartbeatRequestSkipped() unconditionally logs a WARN about "coordinator not known/available" — misleading here since we just logged INFO about an intentional skip. Could we inline the transition + maybeCompleteLeaveInProgress, or guard the WARN inside onHeartbeatRequestSkipped?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3282106530
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1892,10 +1893,21 @@ private void completeShutdown(final boolean cleanRun) {
log.error("Failed to close changelog reader due to the following error:", e);
}
try {
- final GroupMembershipOperation membershipOperation =
- leaveGroupRequested.get() == org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP ? LEAVE_GROUP : REMAIN_IN_GROUP;
- if (membershipOperation == REMAIN_IN_GROUP && streamsRebalanceData.isPresent()) {
- log.info("The consumer will leave the group since the streams group protocol is used");
+ final org.apache.kafka.streams.CloseOptions.GroupMembershipOperation streamsOperation = leaveGroupRequested.get();
+ final GroupMembershipOperation membershipOperation;
+ if (streamsOperation == org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP) {
+ membershipOperation = LEAVE_GROUP;
+ } else if (streamsOperation == org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP) {
+ membershipOperation = REMAIN_IN_GROUP;
+ } else {
+ // DEFAULT: adapt to the active protocol
+ if (streamsRebalanceData.isPresent()) {
+ // Streams protocol: leave the group (consistent with dynamic member behavior)
Review Comment:
Once static is handled in the streams consumer layer (see comment on StreamsGroupHeartbeatRequestManager.shouldSendLeaveHeartbeat), this "consistent with dynamic member behavior" comment becomes misleading — the routing here is intentionally static-agnostic because the streams consumer layer adapts itself. Could you reword along the lines of "Streams protocol: delegate to the consumer, which adapts to static vs dynamic"?
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -489,7 +489,7 @@ private void replaceStreamThread(final Throwable throwable) {
closeToError();
}
final StreamThread deadThread = (StreamThread) Thread.currentThread();
- deadThread.shutdown(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP);
+ deadThread.shutdown(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.DEFAULT);
Review Comment:
Was explicit REMAIN_IN_GROUP, now DEFAULT. With the consumer layer doing the right thing, DEFAULT is reasonable — but the prior code had a clearer intent. Was this intentional ("let the consumer decide") or did it just follow the close() default? A short comment either way would help.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -210,12 +211,20 @@ public int hashCode() {
/**
* If the member is currently leaving the group after a call to {@link #leaveGroup()} or
- * {@link #leaveGroupOnClose()}, this will have a future that will complete when the ongoing leave operation
- * completes (callbacks executed and heartbeat request to leave is sent out). This will be empty if the
- * member is not leaving.
+ * {@link #leaveGroupOnClose(CloseOptions.GroupMembershipOperation)}, this will have a future that will
+ * complete when the ongoing leave operation completes (callbacks executed and heartbeat request to leave
+ * is sent out). This will be empty if the member is not leaving.
*/
private Optional<CompletableFuture<Void>> leaveGroupInProgress = Optional.empty();
+ /**
+ * The operation the member will perform on leaving the group. Remains {@code DEFAULT} until the
+ * member is closing.
+ *
+ * @see CloseOptions.GroupMembershipOperation
+ */
+ private CloseOptions.GroupMembershipOperation leaveGroupOperation = CloseOptions.GroupMembershipOperation.DEFAULT;
Review Comment:
Set in leaveGroupOnClose but never reset. Probably fine because close is terminal, but if we ever rejoin after an aborted close, a stale REMAIN_IN_GROUP will silently suppress the next leave. Could you reset to DEFAULT in transitionToJoining for defense?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -458,20 +463,35 @@ 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. For dynamic members closing with
+ * {@link org.apache.kafka.clients.consumer.CloseOptions.GroupMembershipOperation#REMAIN_IN_GROUP},
+ * the leave heartbeat is skipped and the broker removes the member via session timeout instead.
+ *
+ * @return true if a leave heartbeat should be sent, false otherwise
+ */
+ private boolean shouldSendLeaveHeartbeat() {
Review Comment:
To prepare for static membership under the streams protocol (paralleling what ConsumerHeartbeatRequestManager does for the classic path), this method and the new poll() skip-branch should treat consumer DEFAULT + static the same way they treat REMAIN_IN_GROUP: skip the leave heartbeat. That requires plumbing group.instance.id into StreamsMembershipManager — today groupInstanceId is hard-coded Optional.empty() on line 201, so even a config-provided value never reaches here. Once that's in, StreamThread.completeShutdown can keep passing consumer DEFAULT for streams and the consumer layer decides.
We can't end-to-end test this until static support lands client-side under streams, but the static branches can be unit-tested directly. Without it, the new CloseOptions javadoc promising "Static members: The consumer will remain in the group" under streams will be wrong as soon as static membership reaches the client.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java:
##########
@@ -163,6 +168,84 @@ public void testCloseOptions() throws Exception {
waitForEmptyConsumerGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG), 0);
}
+ @Test
+ public void testCloseOptionsRemainInGroupClassicProtocol() throws Exception {
+ // Classic + REMAIN_IN_GROUP: member must stay in group (no leave heartbeat).
+ // The group should still have a member immediately after close because
+ // the session timeout is set to Integer.MAX_VALUE.
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ assertFalse(isEmptyConsumerGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG)),
+ "Group should still have a member after REMAIN_IN_GROUP close (session timeout is MAX)");
+ }
+
+ @Test
+ public void testCloseOptionsDefaultClassicProtocol() throws Exception {
Review Comment:
before() sets group.instance.id, so this actually exercises Classic + Static + DEFAULT, not the dynamic case. The test would still pass if the Classic branch in StreamThread were flipped, so it doesn't lock the routing down. Could you streamsConfig.remove(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG) here? (Same applies to the pre-existing testCloseOptionsRemainInGroup — just flagging.)
##########
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java:
##########
@@ -1454,11 +1454,15 @@ public CloseOptions leaveGroup(final boolean leaveGroup) {
* Shutdown this {@code KafkaStreams} instance by signaling all the threads to stop, and then wait for them to join.
* This will block until all threads have stopped.
* <p>
- * When using the classic protocol, the consumer will not leave the group explicitly. However, when using
- * the streams group protocol ({@code group.protocol=streams}), the consumer will always leave the group.
+ * Uses {@link org.apache.kafka.streams.CloseOptions.GroupMembershipOperation#DEFAULT DEFAULT} behavior,
Review Comment:
The new bullet "Streams protocol: the consumer leaves the group" doesn't reflect the static case the KIP calls out. Once the streams consumer layer is static-aware, please mention the static-member nuance here (matching the CloseOptions javadoc). Same on the close(Duration) javadoc below.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -539,9 +560,15 @@ public void transitionToFatal() {
*/
public void onHeartbeatRequestSkipped() {
if (state == MemberState.LEAVING) {
- log.warn("Heartbeat to leave group cannot be sent (most probably due to coordinator " +
- "not known/available). Member {} with epoch {} will transition to {}.",
- memberId, memberEpoch, MemberState.UNSUBSCRIBED);
+ if (CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP == leaveGroupOperation) {
Review Comment:
If the coordinator is unavailable during a close, the earlier coordinator.isEmpty() branch in poll() also reaches here, and we'd now log the INFO "skipping because REMAIN_IN_GROUP" rather than the WARN about coordinator availability — the signal that the coordinator was down gets swallowed. Also, once DEFAULT + static is handled, this branch will need to cover that too. Worth passing a reason in, or splitting the two callers.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java:
##########
@@ -163,6 +168,84 @@ public void testCloseOptions() throws Exception {
waitForEmptyConsumerGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG), 0);
}
+ @Test
+ public void testCloseOptionsRemainInGroupClassicProtocol() throws Exception {
+ // Classic + REMAIN_IN_GROUP: member must stay in group (no leave heartbeat).
+ // The group should still have a member immediately after close because
+ // the session timeout is set to Integer.MAX_VALUE.
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ assertFalse(isEmptyConsumerGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG)),
+ "Group should still have a member after REMAIN_IN_GROUP close (session timeout is MAX)");
+ }
+
+ @Test
+ public void testCloseOptionsDefaultClassicProtocol() throws Exception {
+ // Classic + DEFAULT: must behave like REMAIN_IN_GROUP (member stays in group).
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.DEFAULT)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ assertFalse(isEmptyConsumerGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG)),
+ "Group should still have a member after DEFAULT close under Classic protocol");
+ }
+
+ @Test
+ public void testCloseOptionsLeaveGroupStreamsProtocol() throws Exception {
+ // Streams + LEAVE_GROUP: member must leave the group immediately.
+ streamsConfig.remove(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG);
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ waitForEmptyStreamGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG), 0);
+ }
+
+ @Test
+ public void testCloseOptionsDefaultStreamsProtocol() throws Exception {
+ // Streams + DEFAULT: dynamic member must leave the group (consistent with Streams protocol design).
+ streamsConfig.remove(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG);
+ streamsConfig.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name());
+ streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
+
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.DEFAULT)
+ .withTimeout(Duration.ofSeconds(30)));
+
+ waitForEmptyStreamGroup(adminClient, streamsConfig.getProperty(StreamsConfig.APPLICATION_ID_CONFIG), 0);
+ }
+
+ @Test
+ public void testCloseOptionsRemainInGroupStreamsProtocol() throws Exception {
Review Comment:
Nit: comment in this test says "session timeout is set to Integer.MAX_VALUE" but SESSION_TIMEOUT_MS_CONFIG is consumer-side and doesn't apply to streams groups. The test passes because the assertion runs well within the broker default streams-group session timeout. Either fix the comment or set STREAMS_GROUP_SESSION_TIMEOUT_MS_CONFIG via group config so it's robust against default changes.
##########
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()) {
Review Comment:
The "skip the leave heartbeat" condition now appears in three places: isLeavingGroup(), this poll() block, and shouldSendLeaveHeartbeat(). Once DEFAULT + static is added (see other comment), each site grows the same extra condition — which makes consolidating into one shouldSkipLeaveHeartbeat() helper on the manager more attractive. Also, isLeavingGroup() returning false while state is actually LEAVING is a minor footgun; renaming or splitting it would be clearer.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #21579:
URL: https://github.com/apache/kafka/pull/21579#discussion_r3292779944
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java:
##########
@@ -210,12 +211,20 @@ public int hashCode() {
/**
* If the member is currently leaving the group after a call to {@link #leaveGroup()} or
- * {@link #leaveGroupOnClose()}, this will have a future that will complete when the ongoing leave operation
- * completes (callbacks executed and heartbeat request to leave is sent out). This will be empty if the
- * member is not leaving.
+ * {@link #leaveGroupOnClose(CloseOptions.GroupMembershipOperation)}, this will have a future that will
+ * complete when the ongoing leave operation completes (callbacks executed and heartbeat request to leave
+ * is sent out). This will be empty if the member is not leaving.
*/
private Optional<CompletableFuture<Void>> leaveGroupInProgress = Optional.empty();
+ /**
+ * The operation the member will perform on leaving the group. Remains {@code DEFAULT} until the
+ * member is closing.
+ *
+ * @see CloseOptions.GroupMembershipOperation
+ */
+ private CloseOptions.GroupMembershipOperation leaveGroupOperation = CloseOptions.GroupMembershipOperation.DEFAULT;
Review Comment:
Good catch, we should reset to `DEFAULT` for defense
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20167 Introduce CloseOptions.DEFAULT for Kafka Streams [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #21579:
URL: https://github.com/apache/kafka/pull/21579
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org