Spring应用运行时刷新Kafka Producer的方案求助

问题背景 #

当前使用Spring Cloud Stream结合Kafka（Spring Boot 3.0.11、Spring Cloud 2022.0.4），通过StreamBridge发送消息。Kafka Broker要求使用客户端证书，由于已有供Feign Client等组件使用的动态客户端证书（以Base64编码字符串形式存在于应用内），不愿采用静态密钥库文件方式。

已实现自定义SSLContext，证书更新时会同步变更；同时通过自定义SslEngineFactory让Kafka使用该SSLContext。但Feign等其他客户端能正常使用更新后的SSLContext，Kafka却始终保留旧的上下文。尝试通过运行时刷新Kafka Producer来重建并使用新SSLContext，但现有方案均不适用于Spring Cloud Stream场景，求可行的SSL配置更新或Producer刷新方案。

配置示例 #

spring:
  kafka:
    admin:
      client-id: my-client
      fail-fast: true
  cloud:
    stream:
      default-binder: my-binder
      default:
        producer:
          partition-count: 2
      binders:
        my-binder:
          type: kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    binder:
                      autoCreateTopics: false
                      brokers: my-broker:9092
      bindings:
        my-output:
          binder: my-binder
          destination: my-topic
          contentType: application/json
          producer:
            partitionCount: 2
      kafka:
        binder:
          configuration:
            ssl.engine.factory.class: com.example.kafka.ssl.MySslEngineFactory
            security.protocol: SSL
            ssl.endpoint.identification.algorithm: https
            ssl.client.auth: required

消息发送代码示例 #

@Component
public class KafkaService {

    @Autowired
    StreamBridge streamBridge;

    @Autowired
    EventPublishingExecutor eventPublishingExecutor;

    @Value("${spring.cloud.stream.bindings.my-output.destination}")
    String topic;

    public void publishMessage(String message) {
        if (null == streamBridge)
            throw new RuntimeException("Cannot send kafka event: streamBridge is null");

        //Send to broker
        final KafkaPublisherRunnable runnable = new KafkaPublisherRunnable(streamBridge, topic, message);
        eventPublishingExecutor.execute(runnable);
    }
}

可行解决方案 #

方案1：改进自定义SslEngineFactory实现

确保MySslEngineFactory每次创建SslEngine时都获取最新的SSLContext，而非初始化时的缓存：

public class MySslEngineFactory implements SslEngineFactory {
    // 持有动态SSLContext的供应商，确保每次获取最新实例
    private final Supplier<SSLContext> sslContextSupplier;

    public MySslEngineFactory(Supplier<SSLContext> sslContextSupplier) {
        this.sslContextSupplier = sslContextSupplier;
    }

    @Override
    public SSLEngine createSslEngine(String peerHost, int peerPort) {
        SSLContext latestSslContext = sslContextSupplier.get();
        SSLEngine sslEngine = latestSslContext.createSSLEngine(peerHost, peerPort);
        // 配置SSL引擎参数（客户端模式、加密套件等）
        sslEngine.setUseClientMode(true);
        return sslEngine;
    }

    // 实现接口要求的其他方法（如configureSslEngine等）
}

此方式无需刷新Producer，每次建立新连接时都会使用最新证书。

方案2：通过Binder机制手动刷新Producer

利用Spring Cloud Stream的Binder生命周期管理，销毁并重建Producer绑定：

@Autowired
private BinderFactory binderFactory;
@Autowired
private BindingService bindingService;

public void refreshKafkaProducer() {
    // 获取目标binder实例
    Binder<?, ?, ?> binder = binderFactory.getBinder("my-binder", MessageChannel.class);
    if (binder instanceof KafkaMessageChannelBinder kafkaBinder) {
        // 销毁现有Producer绑定
        kafkaBinder.destroyProducerBindings();
        // 触发绑定重新初始化
        bindingService.afterPropertiesSet();
    }
}

注意：该操作会暂时中断消息发送，需配合重试机制保证消息可靠性。

方案3：自定义动态ProducerFactory

实现自定义ProducerFactory，每次创建Producer时注入最新SSL配置：

public class DynamicSslProducerFactory<K, V> extends DefaultKafkaProducerFactory<K, V> {
    private final Supplier<SSLContext> sslContextSupplier;

    public DynamicSslProducerFactory(Map<String, Object> configs, Supplier<SSLContext> sslContextSupplier) {
        super(configs);
        this.sslContextSupplier = sslContextSupplier;
    }

    @Override
    protected Producer<K, V> createProducer(Map<String, Object> configs) {
        // 每次创建Producer时，替换为最新的SSLContext相关配置
        SSLContext latestSslContext = sslContextSupplier.get();
        configs.put(CommonClientConfigs.SSL_CONTEXT_CONFIG, latestSslContext);
        return super.createProducer(configs);
    }
}

随后将该自定义ProducerFactory注入到Kafka Binder的配置中，替换默认实现。

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