强制 Consumer 再均衡

Doubao 中英对照 Enforcing Consumer Rebalance

Kafka clients 现在支持一个触发强制再均衡的选项。从版本 3.1.2 开始，Spring for Apache Kafka 提供了一个通过消息监听器容器在 Kafka consumer 上调用此 API 的选项。调用此 API 时，它只是提醒 Kafka consumer 触发一次强制再均衡；实际的再均衡只会在下一次 poll() 操作时发生。如果已经有一个再均衡正在进行中，调用强制再均衡是一个 NO-OP。调用者必须等待当前再均衡完成后才能再调用另一个。有关详情，请参阅 enforceRebalance 的 javadoc。

以下代码片段展示了使用消息监听器容器强制再均衡的核心。

@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 来获得对消息监听器容器的访问，然后在其上调用 enforceRebalance API。当在监听器容器上调用 enforceRebalance 时，它会将调用委托给底层的 Kafka consumer。Kafka consumer 会在下一次 poll() 操作时触发一次再均衡。