多Binder场景下Spring Cloud Stream Kafka Streams SSL配置重复问题求助

我们的应用监听版本3.4.0的第三方Kafka Broker，配置了5个Kafka Streams订阅：3个共享同一组truststore/keystore文件，另外2个各使用独立的证书文件。启用所有订阅后，共享证书的3个订阅运行正常，另外2个订阅抛出TopicAuthorizationException错误，无法接收消息：

org.apache.kafka.streams.errors.StreamsException: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [topic-2]
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:642)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576)
Caused by: org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [topic-2]

调试后发现，某一个binder的SSL配置被重复应用到所有其他binder上。我们使用的技术栈为：Spring Boot 2.7.5，Spring Cloud Stream Kafka Streams相关依赖如下：

implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka-streams'
implementation 'org.springframework.cloud:spring-cloud-function-context:3.2.8'

这个问题和多binder场景下的JAAS配置冲突逻辑一致，核心原因是多个Kafka Streams binder共享了同一个applicationId，导致配置（包括SSL参数）被相互覆盖。解决方法是为每个binder指定独立的applicationId，确保配置完全隔离：

1. 移除全局的spring.cloud.stream.kafka.streams.binder.applicationId配置
2. 在每个binder的environment配置块内，添加唯一的applicationId，保证每个binder的ID不重复
3. 保留原有SSL配置，确保每个binder的证书文件路径正确

修改后的配置示例：

spring:
  application:
    name: foo-bar-service
  cloud:
    discovery:
      enabled: false
    stream:
      function:
        definition: "${STREAM_FUNCTION_DEFINITION:binder1;binder2;binder3;binder4;binder5}" 
      kafka:
        streams:
          default:
            consumer:
              startOffset: latest
              deserializationExceptionHandler: logAndContinue
      bindings:
        binder1-in-0:
          binder: kafka-1
          contentType: application/json
          destination: ${topic1}
        binder2-in-0:
          binder: kafka-2
          contentType: application/json
          destination: ${topic3}
        binder3-in-0:
          binder: kafka-3
          contentType: application/json
          destination: ${topic2}
        binder4-in-0:
          binder: kafka-4
          contentType: application/json
          destination: ${topic1}
        binder5-in-0:
          binder: kafka-5
          contentType: application/json
          destination: ${topic1}
      binders:
        kafka-1:
          type: kstream
          defaultEnvironment: false
          environment:
            spring.cloud.stream.kafka.streams.binder:
              applicationId: foo-bar-service-binder1
              autoCreateTopics: false
              brokers: broker1.net:9093,broker2.net:9093,broker3.net:9093
              configuration:
                security.protocol: SSL
                ssl.truststore.location: trustore1.jks
                ssl.truststore.password: secret-pass
                ssl.keystore.location: keystore1.jks
                ssl.keystore.password: secret-pass
                ssl.key.password: secret-pass
        kafka-2:
          type: kstream
          defaultEnvironment: false
          environment:
            spring.cloud.stream.kafka.streams.binder:
              applicationId: foo-bar-service-binder2
              autoCreateTopics: false
              brokers: broker1.net:9093,broker2.net:9093,broker3.net:9093
              configuration:
                security.protocol: SSL
                ssl.truststore.location: trustore3.jks
                ssl.truststore.password: secret-pass
                ssl.keystore.location: keystore3.jks
                ssl.keystore.password: secret-pass
                ssl.key.password: secret-pass
        kafka-3:
          type: kstream
          defaultEnvironment: false
          environment:
            spring.cloud.stream.kafka.streams.binder:
              applicationId: foo-bar-service-binder3
              autoCreateTopics: false
              brokers: broker1.net:9093,broker2.net:9093,broker3.net:9093
              configuration:
                security.protocol: SSL
                ssl.truststore.location: trustore2.jks
                ssl.truststore.password: secret-pass
                ssl.keystore.location: keystore2.jks
                ssl.keystore.password: secret-pass
                ssl.key.password: secret-pass
        kafka-4:
          type: kstream
          defaultEnvironment: false
          environment:
            spring.cloud.stream.kafka.streams.binder:
              applicationId: foo-bar-service-binder4
              autoCreateTopics: false
              brokers: broker1.net:9093,broker2.net:9093,broker3.net:9093
              configuration:
                security.protocol: SSL
                ssl.truststore.location: trustore1.jks
                ssl.truststore.password: secret-pass
                ssl.keystore.location: keystore1.jks
                ssl.keystore.password: secret-pass
                ssl.key.password: secret-pass
        kafka-5:
          type: kstream
          defaultEnvironment: false
          environment:
            spring.cloud.stream.kafka.streams.binder:
              applicationId: foo-bar-service-binder5
              autoCreateTopics: false
              brokers: broker1.net:9093,broker2.net:9093,broker3.net:9093
              configuration:
                security.protocol: SSL
                ssl.truststore.location: trustore1.jks
                ssl.truststore.password: secret-pass
                ssl.keystore.location: keystore1.jks
                ssl.keystore.password: secret-pass
                ssl.key.password: secret-pass

Kafka Streams的配置与applicationId强绑定，当多个binder使用同一个applicationId时，Spring Cloud Stream会复用同一个Kafka Streams配置实例，导致后加载的SSL配置覆盖之前的配置，或者某个binder的配置被全局应用。为每个binder分配独立的applicationId后，每个binder会拥有自己独立的配置上下文，SSL证书等参数不会再相互干扰。

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