KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol at Client Side. by chickenchickenlove · Pull Request #22559 · apache/kafka

chickenchickenlove · 2026-06-13T02:33:12Z

Summary

This PR adds the remaining client-side support for static membership
when Kafka Streams uses the streams rebalance protocol
(group.protocol=streams) introduced by KIP-1071.

The change allows group.instance.id to be used with the streams
protocol, sends the proper static-member leave epoch when a static
Streams member closes with DEFAULT or REMAIN_IN_GROUP, and treats
UNRELEASED_INSTANCE_ID as a known fatal heartbeat error.

Some close-related changes that were part of my previous full PR
#21603 are intentionally not
included here because they are already covered by
#21579, which introduced
CloseOptions.DEFAULT for Kafka Streams.

Changes

Allow static membership configuration with group.protocol=streams.
Handle UNRELEASED_INSTANCE_ID explicitly in
StreamsGroupHeartbeatRequestManager.
Add coverage for static member close behavior:
- DEFAULT and REMAIN_IN_GROUP use
  LEAVE_GROUP_STATIC_MEMBER_EPOCH.
- LEAVE_GROUP uses LEAVE_GROUP_MEMBER_EPOCH.
- pollOnClose sends the static leave epoch and instance id.
Update Streams config tests to verify group.instance.id is allowed
for unprefixed, consumer., and
main.consumer. configurations.

Reviewers: Lucas Brutschy lbrutschy@confluent.io

Copilot

Pull request overview

This PR completes client-side support for static membership (group.instance.id) when Kafka Streams uses the Streams rebalance protocol (group.protocol=streams, per KIP-1071). It removes prior config validation that blocked static membership under the Streams protocol, adds explicit fatal handling for UNRELEASED_INSTANCE_ID in the Streams heartbeat manager, and extends unit tests to cover the expected close/leave-epoch behavior for static members.

Changes:

Allow group.instance.id when group.protocol=streams (including unprefixed, consumer., and main.consumer. configs).
Treat UNRELEASED_INSTANCE_ID as a known fatal error in StreamsGroupHeartbeatRequestManager.
Add/extend unit tests for static-member close semantics and poll-on-close request contents.

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java	Replaces the previous “should throw” assertion with a parameterized test validating static membership is accepted under Streams protocol and propagated to main consumer configs.
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java	Removes the Streams-protocol compatibility check that rejected `group.instance.id`, enabling static membership configuration.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java	Adds unit tests ensuring static members use the correct leave epoch on close for DEFAULT/REMAIN_IN_GROUP vs LEAVE_GROUP.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java	Extends fatal-error coverage to include `UNRELEASED_INSTANCE_ID` and verifies poll-on-close includes static leave epoch + instance id for static members.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java	Adds explicit fatal handling for `UNRELEASED_INSTANCE_ID` in error response processing.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

lucasbru

Thanks for the PR @chickenchickenlove !

lucasbru · 2026-06-16T07:44:33Z

+    @Test
+    public void testStaticMemberRemainInGroupUsesStaticLeaveEpochOnClose() {
+        CloseOptions.GroupMembershipOperation operation = CloseOptions.GroupMembershipOperation.REMAIN_IN_GROUP;
+        MemberState expectedState = MemberState.LEAVING;
+        int expectedEpoch = StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+        verifyStaticMemberLeaveOnClose(operation, expectedState, expectedEpoch);
+    }
+
+    @Test
+    public void testStaticMemberDefaultUsesLeaveGroupStaticMemberEpochOnClose() {
+        CloseOptions.GroupMembershipOperation operation = CloseOptions.GroupMembershipOperation.DEFAULT;
+        MemberState expectedState = MemberState.LEAVING;
+        int expectedEpoch = StreamsGroupHeartbeatRequest.LEAVE_GROUP_STATIC_MEMBER_EPOCH;
+        verifyStaticMemberLeaveOnClose(operation, expectedState, expectedEpoch);
+    }
+
+    @Test
+    public void testStaticMemberLeaveGroupUsesLeaveGroupEpochOnClose() {
+        CloseOptions.GroupMembershipOperation operation = CloseOptions.GroupMembershipOperation.LEAVE_GROUP;
+        MemberState expectedState = MemberState.LEAVING;
+        int expectedEpoch = StreamsGroupHeartbeatRequest.LEAVE_GROUP_MEMBER_EPOCH;
+        verifyStaticMemberLeaveOnClose(operation, expectedState, expectedEpoch);
+    }


These three only differ by the operation and expected epoch, could they be a single @ParameterizedTest like testPollOnCloseWhenStaticMemberIsLeaving above? Also the names are inconsistent for the same constant: UsesStaticLeaveEpoch vs UsesLeaveGroupStaticMemberEpoch.

lucasbru · 2026-06-16T07:44:37Z

+        final Metrics localMetrics = new Metrics(time);
+        StreamsMembershipManager membershipManagerWithStaticMember = new StreamsMembershipManager(
+            GROUP_ID,
+            Optional.of("instance-1"),


Minor: this hardcodes "instance-1" while StreamsGroupHeartbeatRequestManagerTest uses an INSTANCE_ID constant. Worth a constant here too for consistency.

lucasbru · 2026-06-16T07:44:41Z

@@ -1600,12 +1600,6 @@ protected StreamsConfig(final Map<?, ?> props,

    private void verifyStreamsProtocolCompatibility(final boolean doLog) {
        if (doLog && isStreamsProtocolEnabled()) {


Now that group.instance.id is accepted with the streams protocol, are there docs that still state static membership isn't supported here? Worth checking the Streams upgrade/config docs so they don't contradict this.

You're right.
There is indeed an issue there as well. I'll address that part too and include it in this PR.

lucasbru · 2026-06-16T07:44:55Z


+    @ParameterizedTest
+    @EnumSource(value = CloseOptions.GroupMembershipOperation.class, names = {"DEFAULT", "REMAIN_IN_GROUP"})
+    public void testPollOnCloseWhenStaticMemberIsLeaving(final CloseOptions.GroupMembershipOperation operation) {


Coverage here is unit-level only. Is there an integration/system test exercising static-member close under group.protocol=streams, or is that expected to come separately?

I was planning to cover this with a Ducktape test.
I'll take a look at the existing integration tests and check whether this can be covered by adding a new test case.

…eams rebalance protocol at Client Side.

chickenchickenlove · 2026-06-22T01:28:23Z

@lucasbru
I've rebased onto latest trunk because CI was failed due to 1password error.
Also, I've addressed them based on your comments.
When you get a chance, please take another look. 🙇‍♂️

lucasbru

Hey. Another round of reviews. Substantiative still. I'm a bit worried about 4.4 at the moment, as I think we are also lacking some testing from the current PRs.

For example consumer has test_fencing_static_consumer, but we don't cover this in the current PRs yet.

There is also testAsyncStaticMemberCloseWithLeaveGroupTriggersRebalance which we'd need to replicate for streams gorups

lucasbru · 2026-06-23T09:00:28Z

                heartbeatRequestState.reset();
                break;

+            case UNRELEASED_INSTANCE_ID:


Nit: the body here is byte-for-byte identical to the INVALID_REQUEST / GROUP_MAX_SIZE_REACHED / STREAMS_INVALID_TOPOLOGY / STREAMS_INVALID_TOPOLOGY_EPOCH / STREAMS_TOPOLOGY_FENCED fall-through group (same logger.error call + handleFatalFailure(error.exception(errorMessage))). UNSUPPORTED_VERSION is standalone because it substitutes UNSUPPORTED_VERSION_ERROR_MESSAGE. Could UNRELEASED_INSTANCE_ID just be added as another label in the fall-through group?

lucasbru · 2026-06-23T09:00:28Z

+                handleFatalFailure(error.exception(errorMessage));
+                break;
+
            case UNSUPPORTED_VERSION:


Now that static membership is enabled, FENCED_INSTANCE_ID becomes reachable — GroupMetadataManager has a throwIfInstanceIdIsFenced overload for Streams. It currently falls to the default branch with a generic 'unexpected error FENCED_INSTANCE_ID' log. ConsumerHeartbeatRequestManager has an explicit case with the message 'This is expected in the case that the member was removed from the group by an admin client, and another member joined using the same group instance id.' Worth adding the same here so operators get an actionable message.

lucasbru · 2026-06-23T09:00:28Z

+        streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP)
+            .withTimeout(Duration.ofSeconds(30)));
+
+        waitForEmptyStreamGroup(


Passing timeoutMs=0 to waitForEmptyStreamGroup means a single-shot check with no retry window: TestUtils.retryOnExceptionWithTimeout sets expectedEnd=now, calls the condition once, and if not met throws immediately. streams.close() returning doesn't guarantee the coordinator has processed the leave heartbeat yet. Under load this will be flaky. Same issue exists in testCloseOptionsLeaveGroupStreamsProtocol and testCloseOptionsDefaultStreamsProtocol in this class.

Should pass DEFAULT_MAX_WAIT_MS or a reasonable explicit timeout here.

lucasbru · 2026-06-23T09:00:28Z

+    @ParameterizedTest
+    @MethodSource("staticMemberLeaveOnCloseOperations")
+    public void testStaticMemberUsesExpectedLeaveEpochOnClose(
+        final CloseOptions.GroupMembershipOperation operation,


localMetrics is never closed. Metrics implements Closeable. The test could close it in a try-finally or just inline the construction into the StreamsMembershipManager constructor call like the other static-member tests in this class do.

github-actions Bot added triage PRs from the community streams consumer clients labels Jun 13, 2026

chickenchickenlove mentioned this pull request Jun 13, 2026

KAFKA-20169: Add ducktape test code for Streams Rebalance Protocol. #22561

Open

lucasbru requested a review from Copilot June 16, 2026 07:32

lucasbru self-assigned this Jun 16, 2026

lucasbru requested a review from squah-confluent June 16, 2026 07:32

lucasbru mentioned this pull request Jun 16, 2026

KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol at Client Side. #21603

Closed

Copilot started reviewing on behalf of lucasbru June 16, 2026 07:33 View session

Copilot AI reviewed Jun 16, 2026

View reviewed changes

lucasbru added the ci-approved label Jun 16, 2026

lucasbru reviewed Jun 16, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label Jun 17, 2026

chickenchickenlove mentioned this pull request Jun 18, 2026

KAFKA-20169: Support static membership for Kafka Streams with the streams rebalance protocol at Server Side. #21565

Closed

chickenchickenlove added 2 commits June 22, 2026 10:11

KAFKA-20169: Support static membership for Kafka Streams with the str…

aa4601a

…eams rebalance protocol at Client Side.

Addressing review.

44d3eb3

chickenchickenlove force-pushed the KAFKA-20169-CLIENT-CORE branch from 5d2ace3 to 44d3eb3 Compare June 22, 2026 01:11

chickenchickenlove requested a review from lucasbru June 22, 2026 01:28

lucasbru reviewed Jun 23, 2026

View reviewed changes

		@@ -1600,12 +1600,6 @@ protected StreamsConfig(final Map<?, ?> props,

		private void verifyStreamsProtocolCompatibility(final boolean doLog) {
		if (doLog && isStreamsProtocolEnabled()) {

Conversation

chickenchickenlove commented Jun 13, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chickenchickenlove commented Jun 22, 2026

Uh oh!

lucasbru left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

chickenchickenlove commented Jun 13, 2026 •

edited by github-actions Bot

Loading