DRAFT KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with GraphNode "keyChangingOperation" #18800

appchemist · 2025-02-04T14:18:28Z

That PR is a POC.

For now, I only focused on modifying KStreamImpl.repartitionRequired to replace it with GraphNode.keyChangingOperation.

So I'll be working on it further to make sure it's the right fix and need to write some test code.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

appchemist · 2025-02-04T14:20:15Z

Hi, @mjsax
If you have a moment, Please take a look

mjsax · 2025-02-05T02:50:09Z

Thanks for this draft -- might take a few days until I find time to take a look.

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java

appchemist · 2025-02-17T12:02:16Z

Hi, @mjsax
After reuse InternalStreamsBuilder#getKeyChangingParentNode(). Some Tests are failed.
So, I'll looking for these failed tests
I haven't had much time lately, so I haven't looked at it much.

After Analyzing the cause, I'll request the review again

However, I would appreciate it if you could take a look at the following comment when you have time.
#18800 (comment)

appchemist · 2025-02-22T06:16:05Z

:streams:testAll now succeed.

@ mjsax
If you have a moment, Please take a look.

If the POC is worth applying, I'll resolve the conflict.

appchemist · 2025-03-05T13:53:24Z

@mjsax kindly ping

mjsax · 2025-06-11T21:31:54Z

@appchemist -- Sorry for dropping the ball on this work. -- I would have time now, to support you with with PR, if you are still interested.

appchemist · 2025-06-12T06:20:21Z

@mjsax Of course, I'm still interested.
Thank you for your time.

… GraphNode "keyChangingOperation"

appchemist · 2025-06-15T10:41:25Z

I've resolve the conflict.
./gradlew clean :streams:testAll is all passed

… GraphNode "keyChangingOperation" - refactoring

mjsax

Thanks. Finally made a pass. Hope the comments/question make sense and help.

mjsax · 2025-06-24T01:05:39Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

-
+    private enum Repartition {
+        NOT_REQUIRED,
+        BY_KEY_ONLY,


Not sure what this means?

The meaning of BY_KEY is that repartitioning is determined based on the keyChangingOperation of the GraphNode.

mjsax · 2025-06-24T01:09:42Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -100,14 +100,14 @@ public <K, V> KStream<K, V> stream(final Collection<String> topics,

        final String name = new NamedInternal(consumed.name()).orElseGenerateWithPrefix(this, KStreamImpl.SOURCE_NAME);
        final StreamSourceNode<K, V> streamSourceNode = new StreamSourceNode<>(name, topics, consumed);
+        streamSourceNode.requireRepartitionByKey();


Not sure what this means? Ie, the name and semantics are not clear to me. When we add a new KStream from a topic, the assumption is, that the KStream is partitioned by key. So this operator does not "require" any repartitioning (it just reads from a topic), and it does also not change the key (so downstream repartitioning is also not required as it's not a key changing operation).

fixed it.
The original intent was streamSourceNode.requireNotRepartition(), but it was incorrectly modified during the refactoring process.

mjsax · 2025-06-24T01:10:02Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -116,14 +116,14 @@ public <K, V> KStream<K, V> stream(final Pattern topicPattern,
                                       final ConsumedInternal<K, V> consumed) {
        final String name = new NamedInternal(consumed.name()).orElseGenerateWithPrefix(this, KStreamImpl.SOURCE_NAME);
        final StreamSourceNode<K, V> streamPatternSourceNode = new StreamSourceNode<>(name, topicPattern, consumed);
+        streamPatternSourceNode.requireRepartitionByKey();


Same question as above.

Same answer as above.

mjsax · 2025-06-24T01:10:59Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java

@@ -614,6 +614,14 @@ private GraphNode getKeyChangingParentNode(final GraphNode repartitionNode) {
        return null;
    }

+    protected boolean isRepartitionRequired(final GraphNode node) {


If we want to know, if node requires repartitioning, why do we add this method here? It seems to belong to GraphNode class?

I agree.
I overlooked it while focusing on reusing the InternalStreamsBuilder.findParentNodeMatching.
I also moved InternalStreamsBuilder.findParentNodeMatching to GraphNode, as it seemed more appropriate there.

mjsax · 2025-06-24T01:12:08Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KGroupedStreamImpl.java

            subTopologySourceNodes,
            name,
            graphNode
        );
    }

+    public boolean repartitionRequired() {
+        return builder.isRepartitionRequired(graphNode);


Cf my commend above. This could be graphNode.repartitionRequired() if we move the method. Or do I miss something?

I Agree.
Fixed it

mjsax · 2025-06-24T01:28:35Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

@@ -93,6 +98,14 @@ public String nodeName() {
        return nodeName;
    }

+    public boolean canResolveRepartition() {
+        return keyChangingOperation || repartition != Repartition.NOT_REQUIRED;


The name of this method and it's logic is not clear to me? What does "resolve" actually mean?

It indicates whether the node is able to determine if repartitioning is needed.
I renamed it to canDetermineRepartition for better clarity.

Also, the original intent was keyChangingOperation || repartition != Repartition.BY_KEY, but it was incorrectly modified during the refactoring process.

mjsax · 2025-06-24T01:29:11Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

+        return keyChangingOperation || repartition != Repartition.NOT_REQUIRED;
+    }
+
+    public boolean isRepartitionRequired() {


nit: rename to repartitioningRequired() or requiresRepartitioning()

renamed the method to repartitioningRequired()

mjsax · 2025-06-24T01:29:57Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

+    }
+
+    public boolean isRepartitionRequired() {
+        return keyChangingOperation || repartition == Repartition.ALWAYS_REQUIRED;


A key-changing operation by itself does not require repartitioning so not clear about the "or" condition. Can you elaborate?

You can find the explanation for why Repartition.REQUIRED is necessary in the linked reference.

mjsax · 2025-06-24T01:31:23Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

@@ -105,6 +118,14 @@ public boolean isMergeNode() {
        return mergeNode;
    }

+    public void requireRepartitionByKey() {
+        this.repartition = Repartition.BY_KEY_ONLY;


nit: avoid unnecessary this.

removed this

mjsax · 2025-06-24T01:31:27Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/GraphNode.java

+    }
+
+    public void requireRepartitionAlways() {
+        this.repartition = Repartition.ALWAYS_REQUIRED;


nit: avoid unnecessary this.

removed this

… GraphNode "keyChangingOperation" - Corrected values in Repartition enum - Refactoring

github-actions bot added triage PRs from the community streams labels Feb 4, 2025

appchemist commented Feb 5, 2025

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java Outdated Show resolved Hide resolved

github-actions bot removed the triage PRs from the community label Feb 6, 2025

github-actions bot added the small Small PRs label Feb 22, 2025

github-actions bot removed the small Small PRs label Feb 26, 2025

KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with…

4ae6934

… GraphNode "keyChangingOperation"

appchemist force-pushed the jira/KAFKA-18692 branch from 37f5762 to 4ae6934 Compare June 15, 2025 10:39

KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with…

cff64b2

… GraphNode "keyChangingOperation" - refactoring

mjsax reviewed Jun 24, 2025

View reviewed changes

KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with…

9fda88f

… GraphNode "keyChangingOperation" - Corrected values in Repartition enum - Refactoring

mjsax mentioned this pull request Sep 3, 2025

KAFKA-19668: processValue() must be declared as value-changing operation #20470

Merged

DRAFT KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with GraphNode "keyChangingOperation" #18800

Are you sure you want to change the base?

DRAFT KAFKA-18692: Consider to unify KStreamImpl "repartitionRequired" with GraphNode "keyChangingOperation" #18800

Uh oh!

Conversation

appchemist commented Feb 4, 2025

Committer Checklist (excluded from commit message)

Uh oh!

appchemist commented Feb 4, 2025

Uh oh!

mjsax commented Feb 5, 2025

Uh oh!

Uh oh!

appchemist commented Feb 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

appchemist commented Feb 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

appchemist commented Mar 5, 2025

Uh oh!

mjsax commented Jun 11, 2025

Uh oh!

appchemist commented Jun 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

appchemist commented Jun 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax 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!

appchemist Jun 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

mjsax Jun 24, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

appchemist Jun 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

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!

Uh oh!

appchemist commented Feb 17, 2025 •

edited

Loading

appchemist commented Feb 22, 2025 •

edited

Loading

appchemist commented Jun 12, 2025 •

edited

Loading

appchemist commented Jun 15, 2025 •

edited

Loading

appchemist Jun 29, 2025 •

edited

Loading

mjsax Jun 24, 2025 •

edited

Loading

appchemist Jun 29, 2025 •

edited

Loading