KStream.repartition explicitly writes a stream to an internal repartition topic and consumes it again. The updated record key selects its partition, so downstream key-based processing receives records on the correct partition. Use it after changing keys when grouping, aggregation, or joining depends on those keys. section 7.2.3 section 7.2.4
The operation adds a sink node and source node to the topology, creating a new sub-topology. Its returned KStream no longer requires repartitioning, so descendant stateful operations avoid further repartition topics unless another key-changing operation occurs. section 7.2.4
Configure repartitioning with a Repartitioned object:
- Set key and value Serdes for records written to the repartition topic. section 7.2.4
- Provide a base topic name. Kafka Streams prefixes it with the application ID and appends -repartition, producing a stable, recognizable internal-topic name. section 7.2.4
- Choose a partition count when redistribution is intended to increase task count, and optionally supply a StreamPartitioner for custom record distribution. section 7.2.4 section 7.2.5
Repartition before repeated stateful work
After selectKey changes a stream’s key, call repartition(Repartitioned.with(stringSerde, stringSerde).withName("proactive")) before groupByKey().count(). The explicit boundary lets the aggregation and a later join use the repartitioned stream without each causing another repartition. section 7.2.4
Repartitioning can also raise available parallelism: more partitions produce more tasks and can allow more stream threads to process records. This helps only when keys are reasonably evenly distributed; a dominant key range can still leave one task carrying most of the work. section 7.2.5
Enable topology optimization to consolidate redundant repartition nodes created around a single key-changing processor. Kafka Streams examines the built topology and retains one necessary repartition node, but does not remove repartitions required after multiple key-changing operations. section 7.2.6
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
.....
Compare the explicit repartition boundary with the aggregation and join that consume its returned KStream. section 7.2.4
