spring cloud stream starter stream kafka consumer SSL configuration

2
votes

Spring cloud stream starter kafka is not loading the configuration when it is wiring the consumer. The following is configuration that I see in my console when I ran it in debug mode:

security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS

I have the following configuration part of bootstrap yml file

    spring:
     cloud:
      stream:
       bindings:
         <binding configuration>
       kafka:
        binder:
         autoCreateTopics: false
         brokers: <list of kafka brokers>
         defaultBrokerPort: <default port>
        configuration:
         security:
          protocol: SSL
         ssl:
          truststore:
           location: <path to cliend truststore jks>
           password: <password>
           type: JKS
          keystore:
           location: <path to cliend keystore jks>
           password: <password>
           type: JKS
          key:
           password: <password>
          enabled:
           protocols: TLSv1.2,TLSv1.1,TLSv1

Can anyone let me know if I am configuring it correctly? I am able to successfully post messages to the topic using spring-kafka producer. I want to make sure that I did it correctly before I consider writing a consumer also in spring kafka.

3
spring-cloud-stream

3 Answers

4
votes

I don't think you can put security and protocol (for example) are at two levels in the sample yaml you provided since Kafka is looking for properties like security.protocol, ssl.truststore.location etc. So, when you create the yaml file, provide all the security related kafka properties at the same level in the hierarchy. Otherwise, spring puts them as key/value pairs. 

spring:
     cloud:
      stream:
       bindings:
         <binding configuration>
       kafka:
        binder:
         autoCreateTopics: false
         brokers: <list of kafka brokers>
         defaultBrokerPort: <default port>
        configuration:
         security.protocol: SSL
         ssl.truststore.location: <path to cliend truststore jks>
         ssl.truststore.password: <password>
         ssl.truststore.type: JKS
...
1
votes

Configuration is loading after changing it to how it is reported in https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/129

Thanks to stackoverflow for recommending the post referring to the above git issue :)

1
votes

Recently I also have undergone through the same problem. Unfortunately sobychacko's answer didn't work for me. Though the approach is correct, I assume the mistake is in the indentation where the "configuration" prop is placed.

Refer to 3.4.1 section in this link - https://docs.spring.io/autorepo/docs/spring-cloud-stream-binder-kafka-docs/1.1.0.M1/reference/htmlsingle/

It worked for me when I placed the "configuration" prop in the same column as spring.cloud.stream.binder.brokers i.e.

spring:   
  cloud:
    stream:
      binder:
        brokers:
        configuration:
         security:
          protocol: SSL
         ssl:
          truststore:
           location: <path to cliend truststore jks>
           password: <password>
           type: JKS
          keystore:
           location: <path to cliend keystore jks>
           password: <password>
           type: JKS
          key:
           password: <password>
          enabled:
           protocols: TLSv1.2,TLSv1.1,TLSv1

With this I can see the corresponding fields filled in the log.