You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "fonsdant (via GitHub)" <gi...@apache.org> on 2025/02/06 00:19:45 UTC
[PR] [WIP] KAFKA-10844: groupBy without shufflin [kafka]
fonsdant opened a new pull request, #18811:
URL: https://github.com/apache/kafka/pull/18811
WIP
--
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: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2763061007
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.
--
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] [WIP] KAFKA-10844: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2716296718
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.
--
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: groupBy without shuffling [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #18811:
URL: https://github.com/apache/kafka/pull/18811#discussion_r2195337504
##########
docs/streams/developer-guide/draft.md:
##########
@@ -0,0 +1,209 @@
+# Optimizing Kafka Streams with skipRepartition
Review Comment:
All right! But before digging into this, I would like to know your opinion on whether it would be worth it. I think this is my first KIP contribution, so I might be paying too much attention to something that won't be that relevant in the end (in this case, a rename).
--
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: groupBy without shuffling [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2759887172
@mjsax, I would like to suggest that we adopt "skip repartition" instead of "mark as partitioned". It seems to me to be more consistent with the rest of the KStream API methods and its functional programming language, because "skip repartition" is actually the action that is performed, while "mark as partitioned" seems to me to be more like a condition to be interpreted ("oh, okay, since this is marked as partitioned, I will _skip repartitioning_") than the action itself. What do you think?
Also, I would like to know if you have other test scenarios in mind.
Thanks in advance!
--
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: groupBy without shuffling [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-3029263087
> Hey @fonsdant, are still planning to work on this? Just checking.
Hi Lucas! First of all, thanks for the review! Sorry for the delayed response. Yes, I would like to continue working on this. Since it's been a while, I'll recap what I've already done in order to address your points. Thanks again! 😃
--
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] [WIP] KAFKA-10844: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2667405773
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.
--
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: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2774353771
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.
--
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: groupBy without shuffling [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #18811:
URL: https://github.com/apache/kafka/pull/18811#discussion_r2195260473
##########
docs/streams/developer-guide/draft.md:
##########
@@ -0,0 +1,209 @@
+# Optimizing Kafka Streams with skipRepartition
+
+## Introduction
+
+Apache Kafka Streams automatically triggers repartitioning when operations change the message key before a stateful
+operation like `groupByKey()`, `aggregate()`, or `count()`. This behavior ensures that data is correctly distributed
+across partitions to guarantee accurate calculations.
+
+However, in many cases, the data is already partitioned correctly, making repartitioning unnecessary and inefficient. To
+address this, we introduce `skipRepartition()`, an API that allows developers to bypass the repartitioning step when it
+is safe to do so, resulting in reduced latency and lower infrastructure costs.
+
+## Motivation
+
+Imagine a streaming e-commerce application where events are partitioned by `customerId`. We want to calculate the total
+amount spent by each customer:
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId) // Already correctly partitioned!
+ .groupByKey() // By default, this triggers a repartition
+ .aggregate(
+ () -> 0,
+ (key, order, total) -> total + order.amount,
+ Materialized.with(Serdes.String(), Serdes.Integer())
+ );
+```
+
+* **The problem:** Even though the stream is already partitioned correctly by customerId, Kafka Streams will create an
+ unnecessary repartition topic, leading to increased latency and resource consumption.
+* **The solution:** We can use `skipRepartition()` to prevent this:
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId)
+ .skipRepartition() // Avoids unnecessary repartitioning
+ .groupByKey()
+ .aggregate(
+ () -> 0,
+ (key, order, total) -> total + order.amount,
+ Materialized.with(Serdes.String(), Serdes.Integer())
+ );
+```
+
+With `skipRepartition()`, Kafka Streams will skip the repartitioning step and process the aggregation directly,
+optimizing performance.
+
+## When NOT to Use skipRepartition
+
+Although `skipRepartition()` is a powerful optimization tool, it should not be used indiscriminately. Here are some
+cases where it must be avoided:
+
+### Stream Joins
+
+Kafka Streams relies on repartitioning during stream joins to align records by key.
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId)
+ .skipRepartition()
+ .join(
+ builder.table("customers-topic"),
+ (order, customer) -> order.amount + customer.discount,
+ Joined.with(Serdes.String(), Serdes.Integer(), Serdes.String())
+ ); // May produce incorrect results!
+```
+
+Joins expect a composite key produced during repartitioning. Skipping this step may cause misaligned records.
+
+### Interactive Queries (IQ)
+
+Interactive queries depend on the partitioning scheme created by repartitioning. Skipping repartitioning can cause query
Review Comment:
Same as above
--
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: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-3379453426
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: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #18811: KAFKA-10844: groupBy without shuffling
URL: https://github.com/apache/kafka/pull/18811
--
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: groupBy without shuffling [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2959924775
Hey @fonsdant, are still planning to work on this? Just checking.
--
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] [WIP] KAFKA-10844: groupBy without shufflin [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-2655354391
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.
--
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: groupBy without shuffling [kafka]
Posted by "fonsdant (via GitHub)" <gi...@apache.org>.
fonsdant commented on code in PR #18811:
URL: https://github.com/apache/kafka/pull/18811#discussion_r2195259197
##########
docs/streams/developer-guide/draft.md:
##########
@@ -0,0 +1,209 @@
+# Optimizing Kafka Streams with skipRepartition
+
+## Introduction
+
+Apache Kafka Streams automatically triggers repartitioning when operations change the message key before a stateful
+operation like `groupByKey()`, `aggregate()`, or `count()`. This behavior ensures that data is correctly distributed
+across partitions to guarantee accurate calculations.
+
+However, in many cases, the data is already partitioned correctly, making repartitioning unnecessary and inefficient. To
+address this, we introduce `skipRepartition()`, an API that allows developers to bypass the repartitioning step when it
+is safe to do so, resulting in reduced latency and lower infrastructure costs.
+
+## Motivation
+
+Imagine a streaming e-commerce application where events are partitioned by `customerId`. We want to calculate the total
+amount spent by each customer:
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId) // Already correctly partitioned!
+ .groupByKey() // By default, this triggers a repartition
+ .aggregate(
+ () -> 0,
+ (key, order, total) -> total + order.amount,
+ Materialized.with(Serdes.String(), Serdes.Integer())
+ );
+```
+
+* **The problem:** Even though the stream is already partitioned correctly by customerId, Kafka Streams will create an
+ unnecessary repartition topic, leading to increased latency and resource consumption.
+* **The solution:** We can use `skipRepartition()` to prevent this:
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId)
+ .skipRepartition() // Avoids unnecessary repartitioning
+ .groupByKey()
+ .aggregate(
+ () -> 0,
+ (key, order, total) -> total + order.amount,
+ Materialized.with(Serdes.String(), Serdes.Integer())
+ );
+```
+
+With `skipRepartition()`, Kafka Streams will skip the repartitioning step and process the aggregation directly,
+optimizing performance.
+
+## When NOT to Use skipRepartition
+
+Although `skipRepartition()` is a powerful optimization tool, it should not be used indiscriminately. Here are some
+cases where it must be avoided:
+
+### Stream Joins
+
+Kafka Streams relies on repartitioning during stream joins to align records by key.
+
+```java
+KStream<String, Order> orders = builder.stream("orders-topic")
+ .selectKey((key, order) -> order.customerId)
+ .skipRepartition()
+ .join(
+ builder.table("customers-topic"),
+ (order, customer) -> order.amount + customer.discount,
+ Joined.with(Serdes.String(), Serdes.Integer(), Serdes.String())
+ ); // May produce incorrect results!
+```
+
+Joins expect a composite key produced during repartitioning. Skipping this step may cause misaligned records.
Review Comment:
Sure! I think this except from KIP can help answering this question:
> The usage of this operation complicates the usage of IQ(Interactive Query) and joins. For reasons that when repartitions happen, records are physically shuffled by a composite key defined in the stateful operation. However, when the repartitions are canceled, records stayed in their original partition by their original key. IQ assumes and uses the composite key instead of the original key. That's when IQ can break downstream. The same applies to joins.
--
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: groupBy without shuffling [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18811:
URL: https://github.com/apache/kafka/pull/18811#issuecomment-3500554764
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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