Spring Kafka使用ABSwitchCluster实现Kafka集群切换不生效问题咨询

切换不生效主要有3处遗漏配置，按以下步骤修改即可：

1. 将ABSwitchCluster绑定到生产者、消费者工厂 #

你仅声明了ABSwitchCluster的Bean，但是没有将其注入到ConsumerFactory和ProducerFactory的bootstrap servers配置中，客户端工厂默认还是读取固定配置的集群地址，不会响应ABSwitchCluster的切换。
修改工厂配置示例：

// 消费者工厂配置
@Bean
public ConsumerFactory<String, Object> consumerFactory(ABSwitchCluster switcher) {
    Map<String, Object> configs = new HashMap<>();
    // 原有其他消费者配置（group.id、反序列化器等）保持不变
    configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, switcher);
    return new DefaultKafkaConsumerFactory<>(configs);
}

// 生产者工厂配置
@Bean
public ProducerFactory<String, Object> producerFactory(ABSwitchCluster switcher) {
    Map<String, Object> configs = new HashMap<>();
    // 原有其他生产者配置（重试次数、序列化器等）保持不变
    configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, switcher);
    return new DefaultKafkaProducerFactory<>(configs);
}

2. 切换时清空客户端缓存 #

Spring Kafka的ConsumerFactory和ProducerFactory默认会缓存已创建的客户端实例，旧实例会持有切换前的集群地址，仅重启监听器不会销毁旧实例。修改切换逻辑：

@RestController
public class ApacheKafkaWebController {

    @Autowired
    ConsumerKakfaConfiguration configuration;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private ABSwitchCluster switcher;
    
    // 注入你项目中使用的KafkaTemplate
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping(value = "/switch")
    public String producer() {
        registry.stop();
        // 清空所有消费者实例缓存
        registry.getListenerContainers().forEach(container -> 
            container.getConsumerFactory().reset()
        );
        // 清空生产者实例缓存
        kafkaTemplate.getProducerFactory().reset();
        // 切换集群
        switcher.secondary();
        registry.start();
        return "switched!";
    }
}

3. 检查版本及全局配置 #

• Spring Kafka版本需≥2.5，低版本对ABSwitchCluster的支持存在缺陷
• 删除配置文件中spring.kafka.bootstrap-servers的固定地址配置，避免覆盖ABSwitchCluster的设置

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