Spring Cloud Stream Kafka Streams 消费者生产者SSL配置不生效问题

核心问题根因 #

• 配置层级错误：你使用的是Kafka Streams binder，普通spring.cloud.stream.kafka.binder.configuration是为普通Kafka消息binder准备的，Streams相关的全局配置需要放到spring.cloud.stream.kafka.streams.binder.configuration路径下，和你已配置的Schema Registry地址同级。
• 自定义binder类型错误：你自定义binder时指定的type为kafka，这是普通Kafka binder的类型，Kafka Streams binder的正确类型是kstream，所以启动时会报找不到kstream类型binder的错误。

解决方案1：直接全局配置（无需自定义binder） #

直接调整配置层级，将SSL配置放到Streams binder的配置项下即可，所有流处理任务会自动共用这套配置，正确配置片段如下：

spring:
  application.name: processingapp
  cloud:
    function.definition: stream1;stream2;stream3
    stream:
      bindings:
        stream1-in-0:
          destination: topic1
        stream1-out-0:
          destination: topic2
        stream2-in-0:
          destination: topic2
        stream2-in-1:
          destination: topic3
        stream2-out-0:
          destination: topic4
        stream3-in-0:
          destination: topic4
        stream3-out-0:
          destination: topic5
      kafka:
        bindings:
          default:
            consumer:
              resetOffsets: false
              startOffset: latest
          stream1-in-0:
            consumer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          stream1-out-0:
            producer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          stream2-in-0:
            consumer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          stream2-in-1:
            consumer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
              materializedAs: sbinfo-outage-mapping-store
          stream2-out-0:
            producer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          stream3-in-0:
            consumer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
          stream3-out-0:
            producer:
              keySerde: io.confluent.kafka.streams.serdes.avro.PrimitiveAvroSerde
              valueSerde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
        streams:
          binder:
            brokers: kafkabrokerurl.com:9092
            configuration: 
              schema.registry.url: https://schemaregistryurl.com
              # SSL配置移到此处即可生效
              security.protocol: SSL
              ssl.truststore.location: /mnt/truststore.jks
              ssl.truststore.type: JKS
              ssl.truststore.password: 你的信任库密码
              ssl.keystore.location: /mnt/keystore.jks
              ssl.keystore.type: JKS
              ssl.keystore.password: 你的密钥库密码
              ssl.key.password: 你的密钥密码（如无则和密钥库密码一致）
              ssl.enabled.protocols: TLSv1.2

注意：你之前的配置中还存在缩进错误，stream节点和function.definition是同级的，不能缩进在function.definition下面，会导致配置加载失败。

解决方案2：自定义命名binder方案 #

如果需要用自定义binder的方式配置，只需修改binder类型为kstream即可，正确配置片段如下：

spring:
  application.name: processingapp
  cloud:
    function.definition: stream1;stream2;stream3
    stream:
      bindings:
        stream1-in-0:
          destination: topic1
          binder: ssl
        stream1-out-0:
          destination: topic2
          binder: ssl
        stream2-in-0:
          destination: topic2
          binder: ssl
        stream2-in-1:
          destination: topic3
          binder: ssl
        stream2-out-0:
          destination: topic4
          binder: ssl
        stream3-in-0:
          destination: topic4
          binder: ssl
        stream3-out-0:
          destination: topic5
          binder: ssl
      binders:
        ssl:
          type: kstream # 此处改为kstream而非kafka
          environment:
            spring:
              cloud:
                stream:
                  kafka:
                    streams:
                      binder:
                        brokers: kafkabrokerurl.com:9092
                        configuration:
                          schema.registry.url: https://schemaregistryurl.com
                          security.protocol: SSL
                          ssl.truststore.location: /mnt/secrets/truststore.jks
                          ssl.truststore.type: JKS
                          ssl.truststore.password: 你的信任库密码
                          ssl.keystore.location: /mnt/secrets/keystore.jks
                          ssl.keystore.type: JKS
                          ssl.keystore.password: 你的密钥库密码
                          ssl.key.password: 你的密钥密码
                          ssl.enabled.protocols: TLSv1.2

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