SASL_SSL SCRAM-SHA-512 broker configuration and JKS truststore for Flink

[jslusher]

I have a Strimzi cluster using Kafka 3.8 and this is the broker listener configuration:

    listeners:
    - name: scram
      authentication:
        type: scram-sha-512
      port: 9096
      type: loadbalancer
      tls: true

All of our clients specify their given client configuration equivalent of the following to connect to these brokers:

                "sasl.mechanism": "SCRAM-SHA-512",
                "sasl.password": config.KAFKA_SASL_PASSWORD,
                "sasl.username": config.KAFKA_SASL_USERNAME,
                "security.protocol": "SASL_SSL",

Basically protocol, mechanism and username/password.

I'm trying to get a Flink jobManager and taskManager to connect using the same parameters:

    'properties.security.protocol' = 'SASL_SSL',
    'properties.sasl.mechanism' = 'SCRAM-SHA-512',
    'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username=\"{KAFKA_SASL_USERNAME}\" password=\"{KAFKA_SASL_PASSWORD}\";',

I'm pretty sure the Flink jobManager is authenticating, or at least I can see in the logs that it sees the topic I'm trying to connect to:

org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator [] - Discovered new partitions: [debezium.sdlc-example.example.my_table-61, debezium.sdlc-example.example.my_table-32, debezium.sdlc-example.example.my_table-65, 

When it spawns a taskManager however, I see the following in the broker logs and the client fails:

23:06:29,132 INFO [SocketServer listenerType=ZK_BROKER, nodeId=2] Failed authentication with /172.26.15.0 (channelId=172.24.152.2:9096-172.26.15.0:8774-864) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-2-ListenerName(SCRAM-9096)-SASL_SSL-8]

I realize that this is likely a Flink problem and not a Strimzi problem per se. It's been suggested to me by a Flink maintainers that I need to specify these parameters:

  'properties.ssl.truststore.location' = '/path/to/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',

I have two questions:

• If I need to specify truststore for my Flink connectors, why do I not need to specify it anywhere else? I'm using several Kafka Connect plugins and in-house Python consumers/producers and nowhere else do I need to supply trusdtstore information
• I want to try setting these parameters to prove whether or not they're necessary. If they are necessary, are these truststore files available by default or is it something I need to set up on the brokers and supply to Flink clients?

[scholzj]

I have no Flink experience and not much Python Kafka experience either, but my take ...

• You seem to be using the Strimzi Cluster CA issued certificates for the TLS encryption (as your listener has no custom broker certificate). In that case, it is definitely expected that all Java applications (which should include Flink) will need the truststore configuration in order to do authenticate the server in the TLS handshake. This should help you to get and configure the truststore: https://strimzi.io/docs/operators/latest/full/deploying.html#configuring-external-clients-to-trust-cluster-ca-str (please keep in mind that the actual Flink configuration might differ, so if the Flink folks told you it should be with the properties. prefix, it might be the case in Flink)
• Different Kafka clients and different languages handle the TLS connections differently. The Apache Kafka Java client always needs to pass the TLS server authentication in order to connect. So you need to configure the truststore. Other clients in other languages might behave differently and might just ignore the TLS server authentication and connect even without the truststore (server cert, CA file, whatever the name in given client is). This would be in general less secure, because it might be easier for an attacker to get your clients to connect to some fake Kafka cluster and either steal your data or supply some invalid data to your application. But otherwise it is technically possible. It could be also that someone configured the certificates at a system level etc. That is hard to say without the exact client and configuration knowledge. But at the end, it is completely possible that Flink needs it while other clients and applications don't.

So I think the advice you got from Flink is valid and you should try to follow it.

[jslusher]

Here is the entirety of our listener configuration:

    listeners:
    - name: scram
      authentication:
        type: scram-sha-512
      port: 9096
      type: loadbalancer
      tls: true
      configuration:
        bootstrap:
          annotations:
            external-dns.alpha.kubernetes.io/hostname: strimzi-kafka-bootstrap.staging.example.com
            external-dns.alpha.kubernetes.io/ttl: "60"
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-type: external
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: ip
        brokers:
        - broker: 0
          advertisedHost: strimzi-kafka-broker-0.staging.example.com
          advertisedPort: 9096
          annotations:
            external-dns.alpha.kubernetes.io/hostname: strimzi-kafka-broker-0.staging.example.com
            external-dns.alpha.kubernetes.io/ttl: "60"
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-type: external
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: ip
        - broker: 1
          advertisedHost: strimzi-kafka-broker-1.staging.example.com
          advertisedPort: 9096
          annotations:
            external-dns.alpha.kubernetes.io/hostname: strimzi-kafka-broker-1.staging.example.com
            external-dns.alpha.kubernetes.io/ttl: "60"
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-type: external
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: ip
        - broker: 2
          advertisedHost: strimzi-kafka-broker-2.staging.example.com
          advertisedPort: 9096
          annotations:
            external-dns.alpha.kubernetes.io/hostname: strimzi-kafka-broker-2.staging.example.com
            external-dns.alpha.kubernetes.io/ttl: "60"
            service.beta.kubernetes.io/aws-load-balancer-scheme: internal
            service.beta.kubernetes.io/aws-load-balancer-type: external
            service.beta.kubernetes.io/aws-load-balancer-nlb-target-type: ip
        brokerCertChainAndKey:
          secretName: strimzi-kafka-tls
          certificate: tls.crt
          key: tls.key
    authorization:
      type: simple

[jslusher]

And then here is the certificate we have configured for the listeners:

apiVersion: cert-manager.io/v1
kind: Certificate
metadata:
  name: strimzi-kafka-sdlc
spec:
  secretName: strimzi-kafka-tls
  issuerRef:
    name: letsencrypt-prod
    kind: ClusterIssuer
    group: cert-manager.io
  dnsNames:
  - strimzi-kafka-bootstrap.staging.example.com
  - strimzi-kafka-broker-0.staging.example.com
  - strimzi-kafka-broker-1.staging.example.com
  - strimzi-kafka-broker-2.staging.example.com

[jslusher]

Here is another view of DEBUG logs, this time with only SSL handshake configured and truststore location settings removed. I see that at first at least there is definitely a connection being made, I think from the jobManager. You can clearly see in the logs that it can see partitions of the topic its configured to source. You can then see a taskManager spawned and that's what seems to cause the errors and fails the job.

[jslusher]

This turned out to be an issue in Flink where I had two connectors configured, but only one of them with authentication credentials. My original connector was working properly the whole time. After setting authentication on the second connector I'm no longer getting SSL errors. @scholzj thank you so much for your guidance on this. I'll mark this as the answer.

[scholzj]

Great, glad you managed to resolve it.