Spring Boot中为spring-kafka配置SSL Bundle实现mTLS遇异常

问题核心 #

你尝试通过SslBundle配置Kafka消费者但连接失败，改用传统SSL密钥库/信任库配置则正常工作。问题出在错误地将Spring Boot的配置属性直接传递给Kafka客户端，而非正确利用Spring Boot的SSL Bundle解析机制。

错误原因 #

spring.kafka.consumer.ssl.bundle是Spring Boot的配置属性，并非Kafka客户端原生支持的参数。你直接将这个键和SslBundle对象放入消费者配置Map中，Kafka客户端无法识别该配置，导致SSL参数未被正确应用，最终引发连接断开报错。

正确配置方式 #

方式1：配置文件自动绑定（推荐） #

直接在application.yml或application.properties中指定SSL Bundle名称，Spring Boot会自动将其转换为Kafka客户端所需的SSL属性：

spring:
  kafka:
    consumer:
      ssl:
        bundle: kafka
      security-protocol: ${你的安全协议，比如SSL或SASL_SSL}

方式2：Java代码手动配置 #

如果需要通过Java Bean自定义配置，可选择两种方式实现：

利用KafkaProperties自动处理

@Bean
public ConsumerFactory<String, Object> consumerFactory(KafkaProperties kafkaProperties, SslBundles sslBundles) {
    // 绑定SSL Bundle到Kafka消费者配置
    kafkaProperties.getConsumer().getSsl().setBundle("kafka");
    kafkaProperties.getConsumer().setSecurityProtocol(kafkaConfiguration.getKafkaSecurityProtocol());
    // 让KafkaProperties构建包含SSL配置的消费者属性
    return new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties(sslBundles));
}

手动从SslBundle提取SSL属性

@Bean
public Map<String, Object> buildConsumerProperties(SslBundles sslBundles) {
    Map<String, Object> configProps = new HashMap<>();
    SslBundle kafkaSslBundle = sslBundles.getBundle("kafka");

    // 配置信任库信息
    configProps.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, kafkaSslBundle.getTruststoreType());
    configProps.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaSslBundle.getTruststore().toUri());
    configProps.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, kafkaSslBundle.getTruststorePassword());

    // 配置密钥库信息（如果存在）
    if (kafkaSslBundle.getKeyStore() != null) {
        configProps.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, kafkaSslBundle.getKeyStoreType());
        configProps.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, kafkaSslBundle.getKeyStore().toUri());
        configProps.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, kafkaSslBundle.getKeyStorePassword());
        // 配置密钥密码（如果存在）
        if (kafkaSslBundle.getKeyPassword() != null) {
            configProps.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, kafkaSslBundle.getKeyPassword());
        }
    }

    configProps.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, kafkaConfiguration.getKafkaSecurityProtocol());
    return configProps;
}

版本兼容性确认 #

你使用的Spring Boot 3.2.2和Spring Kafka 3.1.3完全支持Kafka SSL Bundle功能，不存在功能未适配的问题，只需修正配置方式即可。

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