KAFKA-18193 Refactor Kafka Streams CloseOptions to Fluent API Style#19955

Merged

bbejeck merged 19 commits intoapache:trunkfrom

m1a2st:KAFKA-18193

Oct 7, 2025

Collaborator

m1a2st commented Jun 12, 2025 •

edited by github-actions Bot

Loading

In Kafka Streams, configuration classes typically follow a fluent API
pattern to ensure a consistent and intuitive developer experience.
However, the current implementation of
org.apache.kafka.streams.KafkaStreams$CloseOptions deviates from this
convention by exposing a public constructor, breaking the uniformity
expected across the API.

To address this inconsistency, we propose introducing a new
CloseOptions class that adheres to the fluent API style, replacing the
existing implementation. The new class will retain the existing
timeout(Duration) and leaveGroup(boolean) methods but will enforce
fluent instantiation and configuration. Given the design shift, we will
not maintain backward compatibility with the current class.

This change aligns with the goal of standardizing configuration objects
across Kafka Streams, offering developers a more cohesive and
predictable API.

Reviewers: Bill Bejeckbbejeck@apache.org

m1a2st added 2 commits

June 12, 2025 20:16


          completed the kip

a39a4e0


          update

e86ecec

github-actions Bot added triage streams tools labels

Collaborator Author

m1a2st commented Jun 12, 2025

This patch should wait until the KIP-1153 vote concludes.


          add JacaDoc

c67302c

m1a2st force-pushed the KAFKA-18193 branch from 3b23689 to c67302c Compare

June 12, 2025 14:25

m1a2st added 2 commits

June 12, 2025 22:41


          Merge branch 'trunk' into KAFKA-18193

5988fa6


          fix java document

6d548f0

m1a2st force-pushed the KAFKA-18193 branch from ab1215c to 6d548f0 Compare

June 12, 2025 14:45

m1a2st added 2 commits

June 13, 2025 09:01


          refactor old method

38cb713


          update the modifier

591f5ee

Collaborator Author

m1a2st commented Jun 13, 2025

This KIP has passed. LINK


          Merge branch 'trunk' into KAFKA-18193

eafada8

github-actions Bot commented Jun 20, 2025

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.

github-actions Bot added the needs-attention label


          Merge branch 'trunk' into KAFKA-18193

e335762

github-actions Bot removed the needs-attention label

github-actions Bot commented Jun 28, 2025

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.

github-actions Bot added the needs-attention label


          Merge branch 'trunk' into KAFKA-18193

dcf50b6

github-actions Bot removed the needs-attention label

github-actions Bot commented Jul 16, 2025

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.

github-actions Bot added the needs-attention label


          Merge branch 'trunk' into KAFKA-18193

430a14f

github-actions Bot removed the needs-attention label

github-actions Bot commented Aug 6, 2025

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.

github-actions Bot added the needs-attention label


          Merge branch 'trunk' into KAFKA-18193

348933e

github-actions Bot removed the needs-attention label

github-actions Bot commented Aug 17, 2025

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.

github-actions Bot added the needs-attention label

mjsax mentioned this pull request

KAFKA-18185: remove internal.leave.group.on.close config #19400

Merged

mjsax requested a review from bbejeck

August 29, 2025 22:24

mjsax added kip and removed triage needs-attention labels


          Merge branch 'trunk' into KAFKA-18193

Member

bbejeck commented Oct 3, 2025

@m1a2st can you rebase to fix the merge conflicts? Thanks

Apologies for the delay in the reviewing we'll get this in soon.

m1a2st added 2 commits

October 4, 2025 09:41


          Merge branch 'trunk' into KAFKA-18193

46aa10e

# Conflicts:
#	streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java


          fix the conflict

6b65c0c

bbejeck reviewed

View reviewed changes

Member

bbejeck left a comment

Thanks for the PR @m1a2st overall looks good, I have a couple of comments

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java Outdated

+                   * threads to join.
+                   * @param options  contains timeout to specify how long to wait for the threads to shut down, and a flag
+                   *                 {@link org.apache.kafka.streams.CloseOptions.GroupMembershipOperation#LEAVE_GROUP} to
+                   *                 trigger consumer leave call

Member

bbejeck Oct 4, 2025

Shouldn't this just specify that users can provide either option for remaining in the group or leaving vs. explicity showing LEAVE_GROUP

streams/src/main/java/org/apache/kafka/streams/CloseOptions.java Outdated

+              public class CloseOptions {
+                  /**
+                   * Enum to specify the group membership operation upon leaving group.

Member

bbejeck Oct 4, 2025

I think this should say "upon closing closing the Kafka Streams application" or something similar

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java Outdated

                           // further state reports from the thread since we're shutting down
-                          int numStreamThreads = processStreamThread(streamThread -> streamThread.shutdown(leaveGroup));
+                          int numStreamThreads = processStreamThread(
+                              streamThread -> streamThread.shutdown(operation == org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP)

Member

bbejeck Oct 4, 2025

I'm thinking we could update the StreamThread#shutdown to take a CloseOptions.GroupMembershipOperation parameter, then change the AtomicBoolean to AtomicReference<GroupMembershipOperation> and we can pass the close option directly to the consumer. WDYT?

Collaborator Author

m1a2st Oct 5, 2025

I think updating the StreamThread#shutdown parameter makes sense to me, so I will update the PR.

m1a2st added 2 commits

October 5, 2025 09:10


          fix java document

0914c2a


          updated the shutdown method

5f00221

Collaborator Author

m1a2st commented Oct 5, 2025

Thanks for @bbejeck review, I have updated the PR based on your comments, PTAL

bbejeck reviewed

View reviewed changes

Member

bbejeck left a comment

@m1a2st thanks for the update, just one minor comment to address, then we can get this in.

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java Outdated

                   private final AtomicLong cacheResizeSize = new AtomicLong(-1L);
-                  private final AtomicBoolean leaveGroupRequested = new AtomicBoolean(false);
+                  private final AtomicReference<org.apache.kafka.streams.CloseOptions.GroupMembershipOperation> leaveGroupRequested =
+                      new AtomicReference<>(org.apache.kafka.streams.CloseOptions.GroupMembershipOperation.LEAVE_GROUP);

Member

bbejeck Oct 6, 2025

The default should be REMAIN_IN_GROUP, which corresponds to the default of leaveGroupRequested = false

Collaborator Author

m1a2st Oct 7, 2025

Thanks for pointing that out, that was my mistake.

m1a2st added 2 commits

October 7, 2025 18:01


          addressed by comments

cb08f8f


          Merge branch 'trunk' into KAFKA-18193

10bc44e

bbejeck approved these changes

View reviewed changes

Member

bbejeck left a comment •

edited

Loading

Thanks @m1a2st ! LGTM, I'll merge once the build passes.

bbejeck merged commit ebae768 into apache:trunk

34 of 36 checks passed

Member

bbejeck commented Oct 7, 2025

Merged #19955 into trunk

chia7712 reviewed

View reviewed changes

Member

chia7712 left a comment

@m1a2st sorry for late review. I left a few comments

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-                      final String msgPrefix = prepareMillisCheckFailMsgPrefix(options.timeout, "timeout");
-                      final long timeoutMs = validateMillisecondDuration(options.timeout, msgPrefix);
+                      final CloseOptionsInternal optionsInternal = new CloseOptionsInternal(options);
+                      final String msgPrefix = prepareMillisCheckFailMsgPrefix(optionsInternal.timeout(), "timeout");

Member

chia7712 Oct 7, 2025

Is it safe to pass an Optional<Duration> to prepareMillisCheckFailMsgPrefix? I assume the accepted type is a Duration.

Collaborator Author

m1a2st Oct 7, 2025

It’s safe to pass an Optional object, but we shouldn’t change the error message. Therefore, I’ll update it to use Duration instead.

Invalid value for parameter "timeout" (value was: PT1M). 
Invalid value for parameter "timeout" (value was: Optional[PT1M]).

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-                      final long timeoutMs = validateMillisecondDuration(options.timeout, msgPrefix);
+                      final CloseOptionsInternal optionsInternal = new CloseOptionsInternal(options);
+                      final String msgPrefix = prepareMillisCheckFailMsgPrefix(optionsInternal.timeout(), "timeout");
+                      final long timeoutMs = validateMillisecondDuration(optionsInternal.timeout().get(), msgPrefix);

Member

chia7712 Oct 7, 2025

the timeout could be empty, right?

streams/src/main/java/org/apache/kafka/streams/CloseOptions.java

+                  /**
+                   * Fluent method to set the timeout for the close process.
+                   *
+                   * @param timeout the maximum time to wait for the KafkaStreams to close. If {@code null}, the default timeout will be used.

Member

chia7712 Oct 7, 2025

If {@code null}, the default timeout will be used.

It looks like the implementation is ignoring the comment 😢

Collaborator Author

m1a2st Oct 7, 2025

Yes, I missed this comment. If null values are not allowed, KafkaStreams at L1637 should not be empty.

m1a2st mentioned this pull request

MINOR: Followup KAFKA-18193 for null check and error message #20650

Merged

Collaborator Author

m1a2st commented Oct 7, 2025

Thanks for @chia7712 review, followup PR

chia7712 pushed a commit that referenced this pull request


          MINOR: Followup KAFKA-18193 for null check and error message (#20650)

b1b330e

This PR is a follow-up to KAFKA-18193. It addresses the need for a null
check and an improved error message. Please refer to the previous
comments and the review of
#19955 (review)
for more context.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

eduwercamacaro pushed a commit to littlehorse-enterprises/kafka that referenced this pull request


          KAFKA-18193 Refactor Kafka Streams CloseOptions to Fluent API Style (a…

363a617

…pache#19955)

In Kafka Streams, configuration classes typically follow a fluent API
pattern to ensure a consistent and intuitive developer experience.
However, the current implementation of
`org.apache.kafka.streams.KafkaStreams$CloseOptions` deviates from this
convention by exposing a public constructor, breaking the uniformity
expected across the API.

To address this inconsistency, we propose introducing a new
`CloseOptions` class that adheres to the fluent API style, replacing the
existing implementation. The new class will retain the existing
`timeout(Duration)` and `leaveGroup(boolean)` methods but will enforce
fluent instantiation and configuration. Given the design shift, we will
not maintain backward compatibility with the current class.

This change aligns with the goal of standardizing configuration objects
across Kafka Streams, offering developers a more cohesive and
predictable API.

Reviewers: Bill Bejeck<bbejeck@apache.org>

eduwercamacaro pushed a commit to littlehorse-enterprises/kafka that referenced this pull request


          MINOR: Followup KAFKA-18193 for null check and error message (apache#…

…20650)

This PR is a follow-up to KAFKA-18193. It addresses the need for a null
check and an improved error message. Please refer to the previous
comments and the review of
apache#19955 (review)
for more context.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

shashankhs11 pushed a commit to shashankhs11/kafka that referenced this pull request


          KAFKA-18193 Refactor Kafka Streams CloseOptions to Fluent API Style (a…

652fcf1

…pache#19955)

In Kafka Streams, configuration classes typically follow a fluent API
pattern to ensure a consistent and intuitive developer experience.
However, the current implementation of
`org.apache.kafka.streams.KafkaStreams$CloseOptions` deviates from this
convention by exposing a public constructor, breaking the uniformity
expected across the API.

To address this inconsistency, we propose introducing a new
`CloseOptions` class that adheres to the fluent API style, replacing the
existing implementation. The new class will retain the existing
`timeout(Duration)` and `leaveGroup(boolean)` methods but will enforce
fluent instantiation and configuration. Given the design shift, we will
not maintain backward compatibility with the current class.

This change aligns with the goal of standardizing configuration objects
across Kafka Streams, offering developers a more cohesive and
predictable API.

Reviewers: Bill Bejeck<bbejeck@apache.org>

shashankhs11 pushed a commit to shashankhs11/kafka that referenced this pull request


          MINOR: Followup KAFKA-18193 for null check and error message (apache#…

589821c

…20650)

This PR is a follow-up to KAFKA-18193. It addresses the need for a null
check and an improved error message. Please refer to the previous
comments and the review of
apache#19955 (review)
for more context.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

kip streams tools