You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "zheguang (via GitHub)" <gi...@apache.org> on 2026/05/06 06:33:22 UTC
[PR] KAFKA-10844: KStream mark as partitioned [kafka]
zheguang opened a new pull request, #22221:
URL: https://github.com/apache/kafka/pull/22221
This patch implements KIP-759: Unneeded repartition cancelling. The "unneeded" part is entirely by the user's own judgement :)
A stream with key-changing operation can be marked as partitioned as intended to avoid repartition inserted to its topology. The main use case is for aggregation by a "composite key" -- a stream partitioned by key A can readily aggregate by composite key (A, B) without repartition. The reason being given two records, if their (A, B) are equal, then their A is equal.
Test:
- `markAsPartitioned` is no-op when no key changing
- `markAsPartitioned` should not repartition in resulting topology
- IQ should work when query for the original key intended by `markAsPartitioned`
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22221:
URL: https://github.com/apache/kafka/pull/22221#issuecomment-6307499290
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3213791084
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java:
##########
@@ -170,4 +176,39 @@ private String toRepartitionTopicName(final String input) {
private String repartitionOutputTopic(final Properties props, final String repartitionOperationName) {
return props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG) + "-" + repartitionOperationName + "-repartition";
}
+
+ @Test
+ public void canMarkAsPartitionedWhenKeyNotChanging() {
+ final KStream<Integer, String> notMarked = builder.<Integer, String>stream(inputTopic).mapValues(v -> v + ":" + v);
+ final String topologyNotMarked = builder.build().describe().toString();
+ notMarked.markAsPartitioned();
+ final String topologyMarked = builder.build().describe().toString();
+ assertEquals(topologyNotMarked, topologyMarked);
Review Comment:
May be also assert for notMarked and notMarked.markAsPartitioned() to be same as `this` is returned in the impl
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3216685996
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -526,6 +526,21 @@ <VOut> KStream<K, VOut> flatMapValues(final ValueMapperWithKey<? super K, ? supe
*/
KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
+ /**
+ * Mark the {@code KStream} as partitioned to signal the stream is partitioned as intended,
+ * and does not require further repartitioning in downstream for preceding key-changing operations.
+ *
+ * <p><em>
+ * Note that {@link KStream#markAsPartitioned()} SHOULD NOT be used with interactive query(IQ) or {@link KStream#join}.
+ * For reasons that when repartitions happen, records are physically shuffled by a composite key defined in the stateful operation.
+ * However, if the repartitions were cancelled, records stayed in their original partition by its original key. IQ or joins
+ * assumes and uses the composite key instead of the original key.
+ * </p></em>
Review Comment:
Done. Thanks for spotting!
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3213782942
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -526,6 +526,21 @@ <VOut> KStream<K, VOut> flatMapValues(final ValueMapperWithKey<? super K, ? supe
*/
KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
+ /**
+ * Mark the {@code KStream} as partitioned to signal the stream is partitioned as intended,
+ * and does not require further repartitioning in downstream for preceding key-changing operations.
+ *
+ * <p><em>
+ * Note that {@link KStream#markAsPartitioned()} SHOULD NOT be used with interactive query(IQ) or {@link KStream#join}.
+ * For reasons that when repartitions happen, records are physically shuffled by a composite key defined in the stateful operation.
+ * However, if the repartitions were cancelled, records stayed in their original partition by its original key. IQ or joins
+ * assumes and uses the composite key instead of the original key.
+ * </p></em>
+ *
+ * @return a new, mutated {@code KStream} that will not repartition in subsequent operations.
+ */
+ KStream<K, V> markAsPartitioned();
Review Comment:
Not sure if KStream.scala should also be updated with this ?
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3216698573
##########
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamRepartitionTest.java:
##########
@@ -170,4 +176,39 @@ private String toRepartitionTopicName(final String input) {
private String repartitionOutputTopic(final Properties props, final String repartitionOperationName) {
return props.getProperty(StreamsConfig.APPLICATION_ID_CONFIG) + "-" + repartitionOperationName + "-repartition";
}
+
+ @Test
+ public void canMarkAsPartitionedWhenKeyNotChanging() {
+ final KStream<Integer, String> notMarked = builder.<Integer, String>stream(inputTopic).mapValues(v -> v + ":" + v);
+ final String topologyNotMarked = builder.build().describe().toString();
+ notMarked.markAsPartitioned();
+ final String topologyMarked = builder.build().describe().toString();
+ assertEquals(topologyNotMarked, topologyMarked);
Review Comment:
Hm... leaning towards keeping as is for the spec: "markAsPartitioned is no-op when no key changing". Testing implementation details might be an overkill.
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3213784676
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -526,6 +526,21 @@ <VOut> KStream<K, VOut> flatMapValues(final ValueMapperWithKey<? super K, ? supe
*/
KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
+ /**
+ * Mark the {@code KStream} as partitioned to signal the stream is partitioned as intended,
+ * and does not require further repartitioning in downstream for preceding key-changing operations.
+ *
+ * <p><em>
+ * Note that {@link KStream#markAsPartitioned()} SHOULD NOT be used with interactive query(IQ) or {@link KStream#join}.
+ * For reasons that when repartitions happen, records are physically shuffled by a composite key defined in the stateful operation.
+ * However, if the repartitions were cancelled, records stayed in their original partition by its original key. IQ or joins
+ * assumes and uses the composite key instead of the original key.
+ * </p></em>
Review Comment:
```suggestion
* </em></p>
```
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #22221:
URL: https://github.com/apache/kafka/pull/22221#discussion_r3216691301
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -526,6 +526,21 @@ <VOut> KStream<K, VOut> flatMapValues(final ValueMapperWithKey<? super K, ? supe
*/
KStream<K, V> repartition(final Repartitioned<K, V> repartitioned);
+ /**
+ * Mark the {@code KStream} as partitioned to signal the stream is partitioned as intended,
+ * and does not require further repartitioning in downstream for preceding key-changing operations.
+ *
+ * <p><em>
+ * Note that {@link KStream#markAsPartitioned()} SHOULD NOT be used with interactive query(IQ) or {@link KStream#join}.
+ * For reasons that when repartitions happen, records are physically shuffled by a composite key defined in the stateful operation.
+ * However, if the repartitions were cancelled, records stayed in their original partition by its original key. IQ or joins
+ * assumes and uses the composite key instead of the original key.
+ * </p></em>
+ *
+ * @return a new, mutated {@code KStream} that will not repartition in subsequent operations.
+ */
+ KStream<K, V> markAsPartitioned();
Review Comment:
Saw that KStream.scala is deprecated by 4.3... so maybe not worth updating?
--
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] KAFKA-10844: KStream mark as partitioned [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on PR #22221:
URL: https://github.com/apache/kafka/pull/22221#issuecomment-4491929886
LGTM
Thanks for PR and KIP
--
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