强制消费者重新平衡

Kafka 客户端现在支持一个选项来触发强制重新平衡。从版本 3.1.2 开始，Spring for Apache Kafka 提供了一个选项，可以通过消息监听器容器在 Kafka 消费者上调用此 API。调用此 API 时，它只是提醒 Kafka 消费者触发强制重新平衡；实际的重新平衡只会作为下一个 poll() 操作的一部分发生。如果已经存在重新平衡正在进行中，则调用强制重新平衡将是 NO-OP。调用者必须等待当前重新平衡完成才能调用另一个重新平衡。有关更多详细信息，请参阅 enforceRebalance 的 javadocs。

以下代码片段展示了使用消息监听器容器强制重新平衡的本质。

@KafkaListener(id = "my.id", topics = "my-topic")
void listen(ConsumerRecord<String, String> in) {
    System.out.println("From KafkaListener: " + in);
}

@Bean
public ApplicationRunner runner(KafkaTemplate<String, Object> template, KafkaListenerEndpointRegistry registry) {
    return args -> {
        final MessageListenerContainer listenerContainer = registry.getListenerContainer("my.id");
        System.out.println("Enforcing a rebalance");
        Thread.sleep(5_000);
        listenerContainer.enforceRebalance();
        Thread.sleep(5_000);
    };
}

如上代码所示，应用程序使用 KafkaListenerEndpointRegistry 获取对消息监听器容器的访问权限，然后在该容器上调用 enforceRebalnce API。当在监听器容器上调用 enforceRebalance 时，它会将调用委托给底层的 Kafka 消费者。Kafka 消费者将在下一个 poll() 操作的一部分中触发重新平衡。