Kafka，为什么partition总是0？我是否设置密钥

Question

public void testKafka() throws InterruptedException {
    for (int i = 0; i < 1000; i++) {
        kafkaTemplate.send("topic1","zxcvb", String.valueOf(i));
    }
    Thread.sleep(1000 * 60);
}

    @Component
class KafkaConsumer {
    @KafkaListener(groupId = "01",topics = "topic1")
    public void onMessage1(ConsumerRecord<?, ?> record) {
        System.out.println("1============" + record.topic() + "->" + record.partition() + "->" + record.value() + "============");
    }
    @KafkaListener(groupId = "02", topicPartitions = {
            @TopicPartition(topic = "topic1", partitions = {"2"})
    })
    public void onMessage2(ConsumerRecord<?, ?> record) {
        System.out.println("2============" + record.topic() + "->" + record.partition() + "->" + record.value() + "============");
    }
    @KafkaListener(groupId = "03", topicPartitions = {
            @TopicPartition(topic = "topic1", partitions = {"3"})
    })
    public void onMessage3(ConsumerRecord<?, ?> record) {
        System.out.println("3============" + record.topic() + "->" + record.partition() + "->" + record.value() + "============");
    }
}

以上是我的代码，不知道为什么partition总是0 enter image description here

如果我设置了key，还是0 如果我使用 send (string topic, integer partition, K key, @ nullable V data)

错误：Topic topic1 not present in metadata after 60000 ms.

Answer 1

它将始终将具有相同键的记录发送到同一分区 - 这是设计使然。见 DefaultPartitioner:

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
 * 
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {

该错误可能意味着您正在尝试发送到不存在的分区。

Kafka，为什么partition总是0？我是否设置密钥

Kafka，Why is partition always 0? Whether I set key or not

spring

spring-boot

spring-kafka