如何通过Java Bean而非application.yml配置Spring Kafka SSL？

Spring Kafka 支持通过自定义 SslEngineFactory 注入已构建好的 SSLContext 或 KeyStore，完全不需要依赖本地文件路径。以下是具体实现步骤：

1. 从Credhub加载证书并构建SSL相关Bean #

@Configuration
public class KafkaSslCoreConfig {

    @Value("${kafka.ssl.keystore.base64}")
    private String keystoreBase64;

    @Value("${kafka.ssl.keystore.password}")
    private String keystorePassword;

    @Value("${kafka.ssl.truststore.base64}")
    private String truststoreBase64;

    @Value("${kafka.ssl.truststore.password}")
    private String truststorePassword;

    @Bean
    public KeyStore kafkaKeystore() throws Exception {
        KeyStore keystore = KeyStore.getInstance("JKS");
        byte[] keystoreBytes = Base64.getDecoder().decode(keystoreBase64);
        keystore.load(new ByteArrayInputStream(keystoreBytes), keystorePassword.toCharArray());
        return keystore;
    }

    @Bean
    public KeyStore kafkaTruststore() throws Exception {
        KeyStore truststore = KeyStore.getInstance("JKS");
        byte[] truststoreBytes = Base64.getDecoder().decode(truststoreBase64);
        truststore.load(new ByteArrayInputStream(truststoreBytes), truststorePassword.toCharArray());
        return truststore;
    }

    @Bean
    public SSLContext kafkaSslContext(KeyStore keystore, KeyStore truststore) throws Exception {
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(keystore, keystorePassword.toCharArray());

        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(truststore);

        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);
        return sslContext;
    }
}

2. 自定义SslEngineFactory并配置ConsumerFactory #

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public SslEngineFactory kafkaSslEngineFactory(SSLContext sslContext) {
        DefaultSslEngineFactory sslEngineFactory = new DefaultSslEngineFactory();
        sslEngineFactory.setSslContext(sslContext);
        // 如需配置SSL cipher suites、协议版本等，可在此添加
        return sslEngineFactory;
    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory(KafkaProperties kafkaProperties, SslEngineFactory sslEngineFactory) {
        Map<String, Object> configProps = kafkaProperties.buildConsumerProperties();
        // 移除原配置中依赖文件路径的参数，避免冲突
        configProps.remove("ssl.keystore.location");
        configProps.remove("ssl.truststore.location");
        
        DefaultKafkaConsumerFactory<String, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(configProps);
        consumerFactory.setSslEngineFactory(sslEngineFactory);
        return consumerFactory;
    }

    // 如需配置ProducerFactory，逻辑完全一致
    @Bean
    public ProducerFactory<String, Object> producerFactory(KafkaProperties kafkaProperties, SslEngineFactory sslEngineFactory) {
        Map<String, Object> configProps = kafkaProperties.buildProducerProperties();
        configProps.remove("ssl.keystore.location");
        configProps.remove("ssl.truststore.location");
        
        DefaultKafkaProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(configProps);
        producerFactory.setSslEngineFactory(sslEngineFactory);
        return producerFactory;
    }
}

如果必须通过文件路径配置，最佳实践是在Spring上下文初始化早期生成临时文件，并将路径注入到配置中，确保Kafka自动配置能读取到：

@Component
@Order(Ordered.HIGHEST_PRECEDENCE)
public class SslTempFileGenerator implements ApplicationListener<ApplicationEnvironmentPreparedEvent> {

    @Value("${kafka.ssl.keystore.base64}")
    private String keystoreBase64;

    @Value("${kafka.ssl.truststore.base64}")
    private String truststoreBase64;

    @Override
    public void onApplicationEvent(ApplicationEnvironmentPreparedEvent event) {
        try {
            // 生成临时keystore文件，JVM退出时自动删除
            File keystoreFile = File.createTempFile("kafka-keystore", ".jks");
            keystoreFile.deleteOnExit();
            Files.write(keystoreFile.toPath(), Base64.getDecoder().decode(keystoreBase64));

            // 生成临时truststore文件
            File truststoreFile = File.createTempFile("kafka-truststore", ".jks");
            truststoreFile.deleteOnExit();
            Files.write(truststoreFile.toPath(), Base64.getDecoder().decode(truststoreBase64));

            // 将临时文件路径注入环境，覆盖原有配置
            ConfigurableEnvironment env = event.getEnvironment();
            env.getPropertySources().addFirst(new MapPropertySource("kafka-ssl-temp-files", Map.of(
                "spring.kafka.ssl.keystore.location", keystoreFile.getAbsolutePath(),
                "spring.kafka.ssl.truststore.location", truststoreFile.getAbsolutePath()
            )));
        } catch (Exception e) {
            throw new IllegalStateException("Failed to generate temporary SSL files", e);
        }
    }
}

此方法无需修改ConsumerFactory的配置，Spring Kafka自动配置会读取临时文件路径完成SSL初始化。

从安全运维角度，方案C（流水线注入.jks文件到容器）是更简洁的选择：比如在K8s中用Secret挂载SSL文件到容器指定路径，或在CI/CD阶段将文件复制到容器内，只需配置spring.kafka.ssl.keystore.location指向该路径即可。这种方式避免了代码层面的复杂处理，更符合DevOps最佳实践。

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