如何为EmbeddedKafka监听器设置主题/分区数？测试分区分配异常排查

这个问题我之前也碰到过，核心是对Kafka消费者组的分区分配逻辑和测试工具的用法理解有点偏差，我给你一步步分析：

1. 为什么分区不会全部分配到同一监听器？ #

这是Kafka消费者组的默认行为：

• Kafka的消费者组设计初衷就是让多个消费者共同消费一个主题，通过分区分配策略（默认是Range，也支持RoundRobin等）将分区均衡分散到组内的各个消费者上，避免单消费者负载过高。
• 比如你有3个分区、2个同组消费者，按照Range策略，就会出现一个消费者分到2个分区，另一个分到1个的情况——这完全符合Kafka的集群消费逻辑，不是异常。

2. 测试工具的错误使用点 #

你当前的代码问题出在waitForAssignment的参数传递上：

ContainerTestUtils.waitForAssignment(messageListenerContainer, embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());

这里你给每个监听器容器都传入了主题的总分区数，但实际上每个监听器只会分到总分区的一部分（因为多消费者分摊），所以工具会一直等待“分配到全部分区”，最终超时抛出异常，或者和实际分配数不匹配报错。

3. 针对性解决方案 #

根据你的测试需求，有两种常见的解决思路：

思路一：测试单消费者场景（最简单） #

如果你不需要测试多消费者并发，只需要让所有分区都分配给同一个监听器，只需要把监听器的并发数设为1即可：

@BeforeEach
void setup() {
    // 先把所有监听器的并发数设置为1，确保只有一个消费者实例
    for (MessageListenerContainer container : kafkaListenerEndpointRegistry.getListenerContainers()) {
        if (container instanceof ConcurrentMessageListenerContainer) {
            ((ConcurrentMessageListenerContainer<?, ?>) container).setConcurrency(1);
        }
    }
    // 此时每个监听器（实际只有一个消费者实例）会拿到所有分区，原代码即可正常工作
    for (MessageListenerContainer container : kafkaListenerEndpointRegistry.getListenerContainers()) {
        ContainerTestUtils.waitForAssignment(container, embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());
    }
}

思路二：测试多消费者场景（灵活适配） #

如果需要验证多消费者分摊分区的逻辑，就不能固定等待总分区数，而是要等待所有监听器的已分配分区总数等于主题总分区数：

@BeforeEach
void setup() {
    int totalPartitions = embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic();
    long timeout = 30000; // 30秒超时，可根据情况调整
    long startTime = System.currentTimeMillis();

    // 循环等待所有分区完成分配
    while (System.currentTimeMillis() - startTime < timeout) {
        int assignedTotal = kafkaListenerEndpointRegistry.getListenerContainers().stream()
                .mapToInt(container -> container.getAssignedPartitions().size())
                .sum();
        if (assignedTotal == totalPartitions) {
            break;
        }
        Thread.sleep(100); // 短时间休眠避免循环占用资源
    }
}

这种方式不管有多少个消费者，只要所有分区都被分配出去，就会结束等待，不会出现单个监听器挂起的问题。

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