A StreamPartitioner is a Kafka Streams component that selects the output partition for a record. It becomes especially useful when a windowed aggregation key contains window timestamps: identical underlying keys can otherwise land on different partitions. section 9.1.6
The partitioner receives a topic, a Windowed key, a value, and the partition count, then returns the partition set. A typical implementation serializes windowedKey.key(), hashes those bytes with murmur2, and selects one partition. Null keys or failed serialization return Optional.empty(). section 9.1.6
Partitioning a windowed sensor aggregation
A windowed sensor aggregation can use the original sensor key for placement: implement partitions(), serialize windowedKey.key(), calculate a positive murmur2 hash modulo numPartitions, and return the matching singleton partition. Attach the implementation through Produced.with(...).withStreamPartitioner(windowedStreamsPartitioner) when writing with KStream.to(). section 9.1.6
Listing 9.16 Determining correct partition by the underlying key
@Override
public Optional<Set<Integer>> partitions(String topic,
Windowed<K> windowedKey,
V value,
int numPartitions) {
if(windowedKey == null) {
return Optional.empty();
}
byte[] keyBytes = keySerializer.serialize(topic,
windowedKey.key());
if (keyBytes == null) {
return Optional.empty();
}
Integer partition =
Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
return Optional.of(Collections.singleton(partition));
}
Inspect the partition-selection implementation that restores partitioning by the underlying key. section 9.1.6
Listing 9.17 Adding custom StreamPartitioner to Produced
WindowedStreamsPartitioner<String, IotSensorAggregation>
windowedStreamsPartitioner =
new WindowedStreamsPartitioner<>(stringSerde.serializer());
....
.to(outputTopic, Produced.with(
windowedSerdes, aggregationSerde)
.withStreamPartitioner(windowedStreamsPartitioner));
Inspect how the custom StreamPartitioner is attached to a Produced configuration. section 9.1.6
