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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
