Spring Boot配置指定Kafka Broker后仍连接localhost故障排查

核心问题 #

你配置的Kafka地址未被Spring Boot Kafka自动配置识别，因为使用了自定义配置前缀，而非Spring Boot Kafka约定的标准配置键。

具体分析 #

Spring Boot Kafka自动配置仅读取以spring.kafka为前缀的配置项，比如spring.kafka.bootstrap-servers是指定Broker地址的标准键。你在application.yml中用的config.kafka.servers属于自定义前缀，不会被自动配置类读取，因此应用默认使用Kafka客户端的默认地址localhost:9092。

解决方案 #

有两种可行的解决方式：

方式一：改用标准配置前缀修改application.yml #

直接将配置调整为Spring Boot Kafka的标准格式：

spring:
  kafka:
    bootstrap-servers: "my.server:9092"
    properties:
      security:
        protocol: PLAINTEXT # 对应原配置的ssl: false，PLAINTEXT为非SSL协议

方式二：自定义配置类绑定自定义属性 #

若要保留自定义配置前缀，需手动在配置类中绑定属性并创建消费者相关Bean：

1. 创建配置类绑定自定义属性：

@ConfigurationProperties(prefix = "config.kafka")
public class CustomKafkaProperties {
    private String servers;
    private boolean ssl;

    // Getter & Setter
    public String getServers() {
        return servers;
    }

    public void setServers(String servers) {
        this.servers = servers;
    }

    public boolean isSsl() {
        return ssl;
    }

    public void setSsl(boolean ssl) {
        this.ssl = ssl;
    }
}

2. 修改原KafkaConfig，基于自定义配置创建消费者工厂和监听器容器工厂：

@EnableKafka
@Configuration
@EnableConfigurationProperties(CustomKafkaProperties.class)
class KafkaConfig {

    private final CustomKafkaProperties customKafkaProperties;

    public KafkaConfig(CustomKafkaProperties customKafkaProperties) {
        this.customKafkaProperties = customKafkaProperties;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, customKafkaProperties.getServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my_id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        // 配置SSL协议
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, customKafkaProperties.isSsl() ? "SSL" : "PLAINTEXT");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }

    @KafkaListener(id = "my_id", topics = "some_topic")
    public void listen(String in) {
        // ...
    }
}

验证效果 #

修改完成后重启应用，日志会显示连接my.server:9092而非localhost:9092，同时Kafka管理UI中会出现对应的消费者组。

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