Kafka-Spring) Partition Assignment Based on Producer Key
This post was migrated from Tistory. You can find the original here.
Kafka producer key
1
2
3
4
CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(MSG_TOPIC, key, msg.getMsg());
future.whenComplete((result, ex) -> {
System.out.println("future : " + result.getProducerRecord().value());
});
A producer message may or may not have a key.
If it has one, a hash is computed from that key, so messages with the same key get sent to the same partition.
If it doesn’t, messages seem to be distributed across partitions in a round-robin fashion.
Consumer partition
1
2
3
4
5
6
7
@Bean
public NewTopic messageTopic() {
return TopicBuilder.name(MSG_TOPIC)
.partitions(10)
.replicas(1)
.build();
}
Let’s set up 10 partitions,
1
2
3
4
5
6
@KafkaListener(id = "foo", topics = MSG_TOPIC, clientIdPrefix = "myClientId", topicPartitions =
{ @TopicPartition(topic = MSG_TOPIC, partitionOffsets = @PartitionOffset(partition = "0-9", initialOffset = "0"))}
)
public void listen(ConsumerRecord<Integer, String> record) {
System.out.println(record);
}
and check how the partition ends up being assigned based on the key on the consumer side.
ConsumerRecord(topic = MESSAGE_TOPIC, partition = 3, leaderEpoch = 0, offset = 1, CreateTime = 1693732009669, serialized key size = 4, serialized value size = 4, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103])], isReadOnly = false), key = 4, value = “hi”)
ConsumerRecord(topic = MESSAGE_TOPIC, partition = 3, leaderEpoch = 0, offset = 0, CreateTime = 1693732002044, serialized key size = 4, serialized value size = 4, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103])], isReadOnly = false), key = 2, value = “hi”)
We can see logs like the ones above.
Notice that key=2 and key=4 both end up on the same partition.
Of course, once the number of distinct keys exceeds the number of partitions, it’s expected that different keys will collide on the same partition via hashing — but keep in mind that this can also happen even when the number of keys used is smaller than the number of partitions.
The point of using a key isn’t to give a specific key exclusive ownership of a partition, but rather that messages with the same key can be routed through the same partition.
References
https://devocean.sk.com/blog/techBoardDetail.do?ID=164096
[[Kafka] Loading messages into a specific partition using a partition key
devocean.sk.com](https://devocean.sk.com/blog/techBoardDetail.do?ID=164096)