Spring Kafka中getAssignedPartitions()用法及分区查询相关问题

一、正确使用getAssignedPartitions()方法 #

getAssignedPartitions()是MessageListenerContainer接口的方法，需确保在容器完成分区分配后调用，否则可能返回空集合，具体使用方式如下：

1. 通过注解监听器的注册表获取容器
   如果用@KafkaListener注解定义消费者，可借助KafkaListenerEndpointRegistry获取对应容器实例：

@Autowired
private KafkaListenerEndpointRegistry registry;

public void queryAssignedPartitions() {
    // 替换为你@KafkaListener注解中指定的id值
    MessageListenerContainer container = registry.getListenerContainer("your-listener-id");
    if (container != null) {
        Set<TopicPartition> assignedPartitions = container.getAssignedPartitions();
        for (TopicPartition partition : assignedPartitions) {
            System.out.printf("已分配分区：主题=%s，分区号=%d%n", partition.topic(), partition.partition());
        }
    }
}

建议在容器初始化完成后调用，比如监听ListenerContainerIdleEvent事件，或者在ConsumerAwareListenerErrorHandler中触发。

2. 直接引用手动创建的容器实例
   如果是手动构建ConcurrentMessageListenerContainer这类容器，可直接注入或引用实例调用方法：

@Bean
public ConcurrentMessageListenerContainer<String, String> kafkaListenerContainer() {
    // 完成容器的配置与初始化逻辑
    return container;
}

@Autowired
private ConcurrentMessageListenerContainer<String, String> container;

public void checkAssignedPartitions() {
    Set<TopicPartition> partitions = container.getAssignedPartitions();
    // 处理分区信息
}

二、消息分区信息的获取 #

1. 获取上一条消息的接收分区 #

在消息监听方法中，可直接通过ConsumerRecord对象获取分区详情：

@KafkaListener(id = "your-listener-id", topics = "test-topic")
public void handleMessage(ConsumerRecord<String, String> record) {
    int partition = record.partition();
    String topic = record.topic();
    System.out.printf("当前消息来自：主题=%s，分区=%d%n", topic, partition);
}

无论使用AcknowledgingMessageListener还是ConsumerAwareMessageListener，都能从参数里拿到ConsumerRecord以获取分区信息。

2. 预测下一条消息的发送分区 #

消息发送的分区由**分区器（Partitioner）**决定，默认使用DefaultPartitioner，规则如下：

• 指定分区号：直接发送到目标分区；
• 指定key：通过key的哈希值计算分区；
• 无分区无key：轮询选择分区。

若要提前预测分区，可手动调用分区器的partition方法模拟计算：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public int predictMessagePartition(String topic, String key, String value) {
    ProducerFactory<String, String> producerFactory = kafkaTemplate.getProducerFactory();
    Partitioner partitioner = producerFactory.getConfiguration().get(ProducerConfig.PARTITIONER_CLASS_CONFIG);
    if (partitioner == null) {
        partitioner = new DefaultPartitioner();
    }
    Map<String, Object> configs = producerFactory.getConfiguration();
    return partitioner.partition(
        topic,
        key,
        key.getBytes(StandardCharsets.UTF_8),
        value,
        value.getBytes(StandardCharsets.UTF_8),
        configs
    );
}

注意：该预测仅在生产者配置、分区器逻辑、主题分区数未变更时有效，实际发送时若有外部因素变动，结果可能存在差异。

内容的提问来源于stack exchange，提问作者Sunil Timilsina