You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "liuliuOD (via GitHub)" <gi...@apache.org> on 2026/09/23 12:10:07 UTC
[PR] MINOR: Speed up Streams integration test restarts [kafka]
liuliuOD opened a new pull request, #23564:
URL: https://github.com/apache/kafka/pull/23564
Some Kafka Streams integration tests closed and immediately restarted a Streams instance with the same application ID. With the default shutdown behavior, the consumer could remain in the group until `session.timeout.ms` expired, causing unnecessary waiting before the next instance could start or continue the test scenario.
This change explicitly uses `LEAVE_GROUP` for clean shutdown paths followed by state deletion, restoration, offset manipulation, or application restart. Final teardown paths remain unchanged when group membership removal is not required.
Test time (JUnit XML execution time, excluding Gradle build and compilation):
| Case | Before | After |
|---|---:|---:|
| `LagFetchIntegrationTest` | 56.571s | 12.830s |
| `RestoreIntegrationTest` | 232.285s | 141.425s |
| `SelfManagedOffsetLifecycleIntegrationTest` | 102.714s | 51.617s |
| `SlidingWindowedKStreamIntegrationTest` | 504.080s | 62.447s |
| Total targeted scope (81 tests) | 16m04.97s | 5m41.29s |
Test-only change. No production code, public APIs, test inputs, or assertions are affected.
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #23564:
URL: https://github.com/apache/kafka/pull/23564#discussion_r4087095696
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -257,7 +258,7 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol, final boolean
consumerConfig, outputTopic, initialKeyValues);
// wipe out state store to trigger restore process on restart
- streams.close();
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
Review Comment:
I think leave group is by default in new protocol and this test uses new protocol.
Would this be redundant in this case ?
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "liuliuOD (via GitHub)" <gi...@apache.org>.
liuliuOD commented on code in PR #23564:
URL: https://github.com/apache/kafka/pull/23564#discussion_r4093367362
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -257,7 +258,7 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol, final boolean
consumerConfig, outputTopic, initialKeyValues);
// wipe out state store to trigger restore process on restart
- streams.close();
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
Review Comment:
Since this setting is kept for consistency, I added a comment to clarify its purpose.
b575a3e6904488706db91b08d3a78cb9939dc4f4
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on code in PR #23564:
URL: https://github.com/apache/kafka/pull/23564#discussion_r4090689966
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -257,7 +258,7 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol, final boolean
consumerConfig, outputTopic, initialKeyValues);
// wipe out state store to trigger restore process on restart
- streams.close();
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
Review Comment:
This test covers both protocols, so explicitly setting `LEAVE_GROUP` seems totally fine to me.
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st merged PR #23564:
URL: https://github.com/apache/kafka/pull/23564
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "liuliuOD (via GitHub)" <gi...@apache.org>.
liuliuOD commented on code in PR #23564:
URL: https://github.com/apache/kafka/pull/23564#discussion_r4091556300
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -257,7 +258,7 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol, final boolean
consumerConfig, outputTopic, initialKeyValues);
// wipe out state store to trigger restore process on restart
- streams.close();
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
Review Comment:
I checked the configurations here.
- There is no `group.instance.id`, so the `useNewProtocol = true` cases use the Streams protocol with dynamic membership. In those cases, the default close() already performs `LEAVE_GROUP`.
- For `useNewProtocol = false`, the classic protocol default is to remain in the group.
Would you prefer one of these approaches?
1. Keep explicit `LEAVE_GROUP` only where it is needed for the classic protocol, minimizing redundant calls for the Streams protocol cases.
2. Keep explicit `LEAVE_GROUP` consistently for all restart-related shutdowns, to make the intended lifecycle behavior explicit and uniform.
I’m fine with either approach. I’d like to confirm which style is preferred here.
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #23564:
URL: https://github.com/apache/kafka/pull/23564#discussion_r4096860340
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##########
@@ -257,7 +258,7 @@ public void shouldRestoreNullRecord(final boolean useNewProtocol, final boolean
consumerConfig, outputTopic, initialKeyValues);
// wipe out state store to trigger restore process on restart
- streams.close();
+ streams.close(CloseOptions.groupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
Review Comment:
Looks good. Thanks.
--
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] MINOR: Speed up Streams integration test restarts [kafka]
Posted by "liuliuOD (via GitHub)" <gi...@apache.org>.
liuliuOD commented on PR #23564:
URL: https://github.com/apache/kafka/pull/23564#issuecomment-5835269562
> LGTM — sensible test-only change that should reduce flakiness and speed up restarts.
>
> The change replaces plain `streams.close()` / `streams.close(Duration)` with `close(CloseOptions.groupMembershipOperation(LEAVE_GROUP).withTimeout(...))` across several Streams integration tests. The intent (per the comments) is to make the close path deterministic and faster: explicitly leaving the group avoids waiting for the group coordinator's session timeout / rebalance on close, which is the common source of slow or flaky restarts in these tests.
>
> The change is consistent and mechanical:
>
> * A shared `closeStreamsWithLeaveGroup` helper is introduced where the pattern repeats (LagFetchIntegrationTest, SelfManagedOffsetLifecycleIntegrationTest).
>
> * `tearDown` in SelfManagedOffsetLifecycleIntegrationTest still uses the plain `close(STREAMS_CLOSE_TIMEOUT)` — that's fine, since teardown doesn't need the leave-group optimization.
>
>
> One thing to confirm (non-blocking): `LEAVE_GROUP` changes the semantics of close — the member explicitly leaves the group rather than just closing. For tests that assert on group state or rely on the member staying in the group until session timeout, this could alter behavior. Given these are all restart/restore tests that immediately restart the instance, LEAVE_GROUP is appropriate, but it's worth a quick sanity check that none of them assert on group membership _after_ close.
>
> Also note this is test-only, so no production risk. The `CloseOptions` API usage looks correct (builder pattern with `withTimeout`).
I also checked the affected restart paths and none of them rely on the member remaining in the group after close.
--
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