Co-partitioning aligns join inputs so records with matching keys are available in corresponding partitions. In Kafka Streams, every join participant needs the same partition count and the same key type. This lets tasks process matching records locally rather than looking across unrelated partitions. section 7.3.5 section 8.6
Think of partitioning as the placement rule for a key. Identical keys can be joined only when the topics use compatible placement: with equal partition counts, the same key lands in the same partition on both sides. Equal keys alone are not enough if the topics have different partition counts. section 8.6 section 8.6
Check these conditions before building a join:
- Use the same number of partitions for every required join input. section 7.3.5
- Key both inputs by the same field and key type. section 11.6
- Ensure source-topic producers use the same partitioning class; use an identical StreamPartitioner when writing relevant Kafka Streams sink topics with KStream.to(). section 7.3.5
- Rely on default partitioning strategies when possible, since they avoid separate partitioner coordination. section 7.3.5
Co-partitioning is chiefly a join prerequisite. It applies to KStream–KStream joins, KStream–KTable joins, and KTable–KTable joins. A KStream–GlobalKTable join is different: the GlobalKTable keeps a full copy of records, so it has no co-partitioning requirement. section 8.6 section 8.6
A key-changing operation can break the existing alignment. Kafka Streams marks such a stream as requiring repartitioning and repartitions it before a join when necessary. You can also repartition proactively, selecting a partition count that enables a later join while keeping source topics unchanged. section 7.3 section 7.2.4
Listing 7.10 Changing the key, repartitioning, and performing an aggregation and a join
KStream<String, String> originalStreamOne = builder.stream(...);
KStream<String, String> inputStreamOne =
originalStreamOne.selectKey(...); #1
KStream<String, String> inputStreamTwo = builder.stream(...);
KStream<String, String> repartitioned =
inputStreamOne.repartition(Repartitioned #2
.with(stringSerde, stringSerde)
.withName("proactive"));
repartitioned.groupByKey().count().toStream().to(...); #3
KStream<String, String> joinedStream = inputStreamTwo.join(...) #4
.....
Follow a Java topology that changes a key, explicitly repartitions the stream, then reuses the repartitioned result for an aggregation and a join. section 7.2.4
Joining customer purchases
A retailer and its embedded café publish separate purchase streams keyed by customer ID. When both topics are co-partitioned, a stream–stream join can combine a café purchase and store purchase made within 30 minutes, producing a loyalty-points event. In ksqlDB, streams keyed by custId with equal partition counts can be joined directly on that key. section 7.3 section 11.6
Enriching click events
A clickstream contains a user ID and visited link, while a user table holds current user information. Co-partitioning the stream and table permits a stream–table join that enriches each click event with user details. section 8.6 section 8.6
Partitioning and time solve different parts of a stream join. Co-partitioning determines whether matching keys can meet in the same partition; record timestamps and JoinWindows determine join behavior over time. Timestamp extraction can use broker or producer timestamps, or a custom TimestampExtractor for timestamps embedded in event payloads. section 7.3.5 section 7.3
