SSL handshake failed - Kafka Listener Certificate

[senthil13]

Hi Experts,
I was using the latest strimzi-operator latest version, working fine without any issues. But I have to use the custom certificate since we have our own CA. So First I started following the link and created a cluster with its own CA, which worked fine. As a next step, I did it in two parts.

First part: OWN CA change all certs

Create clients csr + strimzi csr

openssl req -new -newkey rsa:2048 -nodes -keyout strimzi.key -out strimzi.csr -passout pass:"test1234" -subj "/C=US/ST=XXXX/L=XX/O=XXX/OU=XXX/CN=StrimziCA"
openssl req -new -newkey rsa:2048 -nodes -keyout clients.key -out clients.csr -passout pass:"test1234" -subj "/C=US/ST=XXXX/L=XX/O=XXX/OU=XXX/CN=ClientsCA"

Signed both certificates with my own CA, and they provided my certificate + root cert + intermediate cert.

cat strimzi.pem > strimzi-bundle.crt;cat inter.crt >> strimzi-bundle.crt;cat root.crt >> strimzi-bundle.crt
cat clients.pem > clients-bundle.crt;cat inter.crt >> clients-bundle.crt;cat root.crt  >> clients-bundle.crt

Loaded those to the secrets using script

Started the cluster -

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  labels:
    app: my-cluster
spec:
  kafka:
    replicas: 1
    resources:
      requests:
        memory: 2Gi
        cpu: 500m
      limits:
        memory: 2Gi
        cpu: "1"
    jvmOptions:
      -Xms: 1024m
      -Xmx: 1024m
    listeners:
      - name: plain
        port: 9092
        tls: false
        type: internal
      - name: tls
        port: 9093
        tls: true
        type: internal
      - name: external
        port: 9094
        type: ingress
        tls: true
          bootstrap:
            host: my-cluster-bootstrap.tcp.devops.XXX.XXXX.com
            annotations:
              external-dns.alpha.kubernetes.io/hostname: my-cluster-bootstrap.tcp.devops.XXX.XXXX.com.
              external-dns.alpha.kubernetes.io/ttl: "60"
          brokers:
          - broker: 0
            host: my-cluster-broker-0.tcp.devops.XXX.XXXX.com
            annotations:
              external-dns.alpha.kubernetes.io/hostname: my-cluster-broker-0.tcp.devops.XXX.XXXX.com.
              external-dns.alpha.kubernetes.io/ttl: "60"
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    resources:
      requests:
        memory: 1Gi
        cpu: "0.3"
      limits:
        memory: 1Gi
        cpu: "0.5"
    storage:
      type: ephemeral
  clusterCa:
    generateCertificateAuthority: false
  clientsCa:
    generateCertificateAuthority: false
  entityOperator:
    tlsSidecar:
      resources:
        requests:
          cpu: 200m
          memory: 64Mi
        limits:
          cpu: 500m
          memory: 128Mi
    topicOperator:
      watchedNamespace: kafka
      reconciliationIntervalSeconds: 60
      logging:
        type: inline
        loggers:
          rootLogger.level: INFO
      resources:
        requests:
          cpu: 500m
          memory: 512Mi
        limits:
          cpu: 1
          memory: 1Gi

but the operator didn't create the zookeeper node.

Control key when opening logs to launch a new window.
Detected Zookeeper ID 1
Preparing truststore
Adding /opt/kafka/cluster-ca-certs/ca.crt to truststore /tmp/zookeeper/cluster.truststore.p12 with alias ca
Certificate was added to keystore
Preparing truststore is complete
Looking for the right CA
No CA found. Thus exiting.

Second part: Kafka Listener certificate

Since the first part failed I want to try the second part, so that operator can handle the internal certificate and outside consumer/producer. I used the same YAML just commenting below lines. named it as name: strimzi-cluster

  clusterCa:
    generateCertificateAuthority: false
  clientsCa:
    generateCertificateAuthority: false

Created the certificate with SAN.

openssl genrsa -out strimzi-cluster-private-cert.key 2048
openssl req -new -key strimzi-cluster-private-cert.key -out strimzi-cluster-private-cert.csr -subj "/C=US/ST=XXXX/L=XX/O=XXX/OU=XXX/CN=StrimziCA" -addext "subjectAltName = DNS.1:*.strimzi-cluster-kafka-brokers,DNS.2:*.strimzi-cluster-kafka-brokers.kafka.svc,DNS.3:strimzi-cluster-kafka-bootstrap,DNS.4:strimzi-cluster-kafka-bootstrap.kafka.svc"

Signed from CA.

openssl pkcs8 -topk8 -inform PEM -outform PEM -nocrypt -in strimzi-cluster-private-cert.key -out strimzi-cluster-private-pkcs8.key

added the root and intermediate certificates.

cat strimzi-cluster-private-cert.csr > strimzi-bundle.crt;cat inter.crt >> strimzi-bundle.crt;cat root.crt >> strimzi-bundle.crt
kubectl create secret generic strimzi-cluster-private-cert --from-file=ca.crt=strimzi-cluster-private-cert.crt --from-file=ca.key=strimzi-cluster-private-pkcs8.key -n kafka

added below lines and updated.

        configuration:
          brokerCertChainAndKey:
            secretName: strimzi-cluster-private-cert
            certificate: ca.crt
            key: ca.key

kubectl apply -f kafka-ephemeral.yaml -n kafka

Added root and intermediate certificates to the trust store.

keytool -import -trustcacerts -alias CARoot -file root.crt -keystore kafka.client.truststore.jks -storepass test1234 -noprompt
keytool -import -trustcacerts -alias CAIntermediate -file inter.crt -keystore kafka.client.truststore.jks -storepass test1234 -noprompt

kafka-console-producer --broker-list strimzi-cluster-bootstrap.tcp.devops.XXX.XXXX.com:443 --producer-property security.protocol=SSL --producer-property ssl.truststore.password=test1234 --producer-property ssl.truststore.location=kafka.client.truststore.jks  --topic test-topic

Getting, SSL handshake

Broker Logs
2022-10-10 19:58:37,772 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.11.0 (channelId=10.42.7.35:9094-10.42.11.0:51069-3) (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SSL-11]

Client logs
[2022-10-10 15:58:11,955] DEBUG [Producer clientId=console-producer] Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 102400, SO_TIMEOUT = 0 to node -1 (org.apache.kafka.common.network.Selector)
[2022-10-10 15:58:12,551] DEBUG [Producer clientId=console-producer] Completed connection to node -1. Fetching API versions. (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,599] INFO [Producer clientId=console-producer] Failed authentication with strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com/XXXX (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2022-10-10 15:58:12,600] DEBUG [Producer clientId=console-producer] Node -1 disconnected. (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,601] ERROR [Producer clientId=console-producer] Connection to node -1 (strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com/XXXX:443) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,601] WARN [Producer clientId=console-producer] Bootstrap broker strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,701] DEBUG [Producer clientId=console-producer] Initialize connection to node strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 (id: -1 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,702] DEBUG [Producer clientId=console-producer] Initiating connection to node strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 (id: -1 rack: null) using address strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com/XXXX (org.apache.kafka.clients.NetworkClient)
[2022-10-10 15:58:12,725] DEBUG [Producer clientId=console-producer] Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 102400, SO_TIMEOUT = 0 to node -1 (org.apache.kafka.common.network.Selector)

Question

• What is the best way to use our own CA?
• Is it using own CA for all certificates or Kafka listen alone?
• Documents show how to create the custom CA, but not showing how to connect.
• Since we do have external clients we cant share the server certificate, but clients can sign their own cert with the same CA.

Thanks
Senthil

[scholzj]

First of all, let's clarify what exactly you mean with latest version. The container images and installation files with the tag :latest are from the main branch. These are unstable since these are under development. What you want to be using is the latest release which is currently 0.31.1. It is not really clear what do you mean with latest here, so I wanted to clarify it.

First part: OWN CA change all certs

Sorry, but from your description it is not really clear what you did. Please follow the steps described here: https://strimzi.io/docs/operators/latest/full/configuring.html#installing-your-own-ca-certificates-str. If you want to move from Strimzi managed CA to your own CA, you should first configure the Kafka CR to tell the operator that you want to now use your own CA (this is the generateCertificateAuthority: false part). And then follow the own CA renewal process to roll out your new CA: https://strimzi.io/docs/operators/latest/full/configuring.html#renewing-your-own-ca-certificates-str

Also, keep in mind that the CA needs to be a CA => the certificate you provide needs to be marked as usable for CA purposes (but it can be just an intermediate CA, does not have to be a Root CA).

From the error, it seems clear that you didn't rolled it correctly. If you are starting with a fresh cluster, you do not need to do the renewal, just do your own CA from the start.

Second part: Kafka Listener certificate

I think that in most cases you will use either your own CA or the listener certificate. The difference is that when using your own CA, you need to provide a CA as mentioned above. And this CA is used to sign all certificates used to secure the Kafka cluster. So it is not just used for the communication between your Kafka clients and brokers. It is also used for communication between the brokers, between ZooKeeper nodes, between Kafka and ZooKeeper. So it is used for more things, but it is also more complicated.

The listener certificate is much easier. You provide a server certificate which will be used for a particular listener in the Kafka broker. It does not need to be CA as it is not used to sign any additional certificates. It will be used only for communication between your clients and your Kafka brokers.

[scholzj]

Are you sure your CAs are marked as CAs? I do not see it anywhere in the OpenSSL commands (but I don't remember how exactly it is set with OpenSSL): All CAs in the chain should be configured using the X509v3 Basic Constraints extension. Basic Constraints limit the path length of a certificate chain.

[scholzj]

Oh, one more comment => you definitely should not be deleting any secrets already in use by the cluster. That will of course not work. As I said on the beginning, if you have existing cluster, you need to follow the renewal steps. You should also not touch the Clients CA secrets if you are not configuring the CR to use your own Clients CA.

PS: This is my test repository using CFSSL instead of OpenSSL if you want to try it: https://github.com/scholzj/strimzi-custom-ca-test

[senthil13]

Yes that's a CA, I verified with normal CA validation openssl command.

openssl req -new -x509 -keyout ca-key.pem -out ca.pem -days 365 -passout pass:"test1234" -subj "/C=US/ST=XXXX/L=Oaks/O=XXXX/OU=XXXX/CN=RootCA"

Yes, I am deleting all secrets. Like below. After that I am even verifying with Kubectl command before creating any new.

kubectl -n kafka delete secret strimzi-cluster-clients-ca-cert
kubectl -n kafka delete secret strimzi-cluster-cluster-ca-cert
 
kubectl -n kafka delete secret strimzi-cluster-clients-ca
kubectl -n kafka delete secret strimzi-cluster-cluster-ca

If you see my initial notes, there I mentioned your test repo. By using your test repo its working fine. But the instruction is completely different. For example in your test repo you create only ca.crt without ca.p12 and password. So I was following the current documents. https://strimzi.io/docs/operators/latest/full/configuring.html#installing-your-own-ca-certificates-str

[senthil13]

I think it's hitting this loop and unable to find the CA there.

CA=$(find_ca "/opt/kafka/cluster-ca-certs" "/opt/kafka/zookeeper-node-certs/$HOSTNAME.crt")

if [ ! -f "$CA" ]; then
    echo "No CA found. Thus exiting."
   exit 1
fi

Since the zookeeper pods are not coming to live I am unable to shell into the pod and check for the location.

[scholzj]

Yes that's a CA, I verified with normal CA validation openssl command.

What commands? You are not passing the CA flags anywhere.

Yes, I am deleting all secrets. Like below. After that I am even verifying with Kubectl command before creating any new.

But you cannot delete the secrets if you have some existing cluster. These are needed until the new CA is rolled out.

[senthil13]

What commands? You are not passing the CA flags anywhere.

Am I missing any falgs? I use below command to verify or this https://www.sslshopper.com/certificate-decoder.html through the browser.

openssl x509 -in ca.pem -text -noout

Certificate:
    Data:
        Version: 3 (0x2)
        Serial Number:
            14:dd:80:e8:67:09:93:07:8e:a3:cc:42:4c:82:e3:bc:27:20:35:ce
        Signature Algorithm: sha256WithRSAEncryption
        Issuer: C = US, ST = XXXX, L = XXXX, O = XXXX, OU = XXXX, CN = RootCA
        Validity
            Not Before: Oct 11 15:44:28 2022 GMT
            Not After : Oct 11 15:44:28 2023 GMT
        Subject: C = US, ST = XXXX, L = XXXX, O = XXXX, OU = XXXX, CN = RootCA
        Subject Public Key Info:
            Public Key Algorithm: rsaEncryption
                RSA Public-Key: (2048 bit)
                Modulus:
                    00:d4:f9:ac:cd:a0:1f:0c:67:ba:c1:8a:84:4d:b2:
                    4a:bb:ee:b1:d8:69:a5:4b:45:e1:3e:3f:ea:96:66:
                    19:82:40:06:fe:aa:c3:ff:6f:8a:22:0f:5f:e0:05:
                    6b:bf:75:f3:4d:8b:9d:e2:b2:c5:f1:6d:2c:b9:db:
                    b3:fb:a7:b2:87:dd:9d:4f:d7:81:96:bb:21:d0:18:
                    d7:5c:ce:af:2f:1c:53:7b:f3:71:e2:a8:6c:80:b2:
                    ec:9b:5a:f8:8a:37:95:2c:ed:13:d9:41:de:b2:d4:
                    a3:26:88:35:11:b7:4b:65:cf:50:49:78:02:af:0c:
                    db:2b:16:8c:f2:07:f9:25:4d:bf:11:87:4e:a7:b9:
                    9d:02:e3:c2:88:b7:8f:45:1a:d5:51:52:b8:1c:b3:
                    f7:66:1c:41:0d:f8:aa:3b:9d:f9:5c:27:17:7d:ea:
                    b5:a9:48:83:c7:33:46:c3:a6:25:bf:c3:6a:fc:9e:
                    20:de:62:a8:8f:1b:db:2e:f1:00:11:f4:8b:45:94:
                    65:2e:ec:bf:43:4a:0c:a1:24:c0:bc:45:e3:34:28:
                    06:16:22:84:97:7f:41:58:c6:f2:4c:f2:12:e6:a0:
                    7d:59:cf:7b:28:5a:34:88:cf:cc:61:7b:5c:14:46:
                    54:22:d2:15:25:aa:7b:10:c3:08:b7:99:35:a1:6d:
                    d5:f7
                Exponent: 65537 (0x10001)
        X509v3 extensions:
            X509v3 Subject Key Identifier:
                74:E9:39:0E:A4:CD:78:C4:96:C5:39:35:0A:07:E6:9E:5E:7D:C9:E3
            X509v3 Authority Key Identifier:
                keyid:74:E9:39:0E:A4:CD:78:C4:96:C5:39:35:0A:07:E6:9E:5E:7D:C9:E3

            X509v3 Basic Constraints: critical
                CA:TRUE
    Signature Algorithm: sha256WithRSAEncryption
         bc:c5:12:8c:4f:2d:dd:97:57:c0:6f:97:52:8c:1d:3e:13:b3:
         be:50:1d:33:96:71:35:b0:e9:74:00:40:3c:26:95:af:3a:19:
         39:5c:b7:15:a1:7b:11:13:3e:8c:bb:89:fb:e1:94:8a:06:bb:
         d4:92:d5:06:c9:7e:07:02:3e:63:e0:f4:fa:24:b1:b6:7e:67:
         30:16:43:1f:91:11:3b:b4:05:03:0e:ba:30:f2:d5:99:ac:c0:
         9d:5d:09:cf:6e:dd:e2:8f:8e:b9:c5:69:07:e1:30:17:66:85:
         e8:3b:0c:f6:01:3d:32:46:b8:29:15:33:58:e1:f1:10:84:f4:
         17:f3:71:fd:8d:9d:04:42:4f:67:f2:c7:c0:ef:9d:a0:4c:6a:
         9a:59:99:e7:16:66:b8:b8:e8:02:5d:33:df:e4:a0:73:14:fc:
         f3:3a:10:06:1b:fe:fe:eb:6c:21:50:54:ed:b9:2a:66:0d:7d:
         bb:fc:6d:88:a8:df:d7:7c:ec:7d:cd:43:54:56:c3:6e:1b:cd:
         d1:22:ea:0d:6a:25:c9:bb:20:10:9b:5f:88:9f:bf:88:22:7e:
         65:5c:91:62:70:5d:8c:b0:4b:d4:e1:a4:fb:d1:39:c9:f2:4f:
         8f:13:39:60:b1:63:b3:17:ea:6a:64:b4:6c:18:01:f3:cc:cc:
         0a:24:9d:fe

But you cannot delete the secrets if you have some existing cluster. These are needed until the new CA is rolled out.

Yes, I am careful,, I am creating a new cluster only, not modifying the existing cluster.

[scholzj]

Then why do you need to delete the secrets if you create a new cluster?

You have the CA flag in the Root CA. If you have it in the other CAs it should be fine. I don't know why it doesn't work. You will need to diff it against what the CFSSL in my demo generates 🤷

[senthil13]

Then why do you need to delete the secrets if you create a new cluster?

I am just following the docs. The output will be like the below.

Error from server (NotFound): secrets "strimzi-cluster-cluster-ca-cert" not found
secret/strimzi-cluster-cluster-ca-cert created
secret/strimzi-cluster-cluster-ca-cert labeled
secret/strimzi-cluster-cluster-ca-cert annotated
Error from server (NotFound): secrets "strimzi-cluster-cluster-ca" not found
secret/strimzi-cluster-cluster-ca created
secret/strimzi-cluster-cluster-ca labeled
secret/strimzi-cluster-cluster-ca annotated
Error from server (NotFound): secrets "strimzi-cluster-clients-ca-cert" not found
secret/strimzi-cluster-clients-ca-cert created
secret/strimzi-cluster-clients-ca-cert labeled
secret/strimzi-cluster-clients-ca-cert annotated
Error from server (NotFound): secrets "strimzi-cluster-clients-ca" not found
secret/strimzi-cluster-clients-ca created
secret/strimzi-cluster-clients-ca labeled
secret/strimzi-cluster-clients-ca annotated

One thing I am trying now, Not sure if that is the correct way or not, but it works.

only created client ca, not the cluster ca.

Create CSR only for Clients
openssl req -new -newkey rsa:2048 -nodes -keyout clients.key -out clients.csr -passout pass:"test1234" -subj "/C=US/ST=XXXX/L=Oaks/O=XXXX/OU=XXXX/CN=ClientsCA"

Signed with External CA and created the bundle
cat clients.pem > clients-bundle.crt;cat inter.crt >> clients-bundle.crt;cat root.txt >> clients-bundle.crt

Added below in the kafak CR.

  clientsCa:
    generateCertificateAuthority: false

Create the cluster, all pods came fine. Now to connect, I followed like below.

kubectl get secret strimzi-cluster-clients-ca-cert -n kafka -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
keytool -import -trustcacerts -alias root -file ca.crt -keystore truststore.jks -storepass test1234 -noprompt

kafka-console-producer --broker-list strimzi-cluster-bootstrap.tcp.devops.XXX.XXXX.com:443 --producer-property security.protocol=SSL --producer-property ssl.truststore.password=test1234 --producer-property ssl.truststore.location=./truststore.jks --topic test-topic

kafka-console-consumer --bootstrap-server strimzi-cluster-bootstrap.tcp.devops.XXX.XXX.com:443 --consumer-property security.protocol=SSL --consumer-property ssl.truststore.password=test1234 --consumer-property ssl.truststore.location=./truststore.jks --topic test-topic --from-beginning

demo generates

Unfortunately, I can't follow them because as I mentioned already the steps are different from the documents.

Will try some other try ways tomorrow.

~thanks

[scholzj]

Well, the Clients CA is used for something completely different than the Cluster CA. So even if it might be wrong, it would not stop the deployment. And since you didn't configured TLS client authentication, it is actually not used anywhere. But the Clients CA does not belong to the truststore on the client. It has nothing to do with server authentication.

[senthil13]

Ok, thanks for the information. I may give it one more day to try this OWN CA and then move it to the listener certificate side.

The main thing that we looking for is, we can't provide Strimizi CA signed cert to outside Kafka consumers/producers who are coming outside of our firm. They want a cert that is signed by Trustware or Digicert - External CA. So we need to have a cert on Kafka servers which is signed by an External CA.

~ thanks

[scholzj]

Well, as I told you from the beginning, this is what the custom listener certificate does. I doubt some trusted CA will sign a CA certificate for you. They would normally sign only a server certificate anyway.

[senthil13]

@scholzj - A big thanks.

As mentioned earlier, I tried today with the Kafka listener certificate and I am able to connect without any issues. I am going to document this here so that anyone facing the same issue can refer here.

Create a CSR and key for the listener.

openssl req -new -newkey rsa:2048 -nodes -keyout listener.key -out listener.csr -passout pass:"test1234" -subj "/C=US/ST=XXXX/L=XXXX/O=XXXX/OU=XXXX/CN=listenerCA" -addext "subjectAltName = DNS:*.strimzi-cluster-kafka-brokers,DNS:*.strimzi-cluster-kafka-brokers.kafka.svc,DNS:strimzi-cluster-kafka-bootstrap,DNS:strimzi-cluster-kafka-bootstrap.kafka.svc,DNS:strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com,DNS:strimzi-cluster-broker-0.tcp.devops.XXXX.XXXX.com,DNS:strimzi-cluster-broker-1.tcp.devops.XXXX.XXXX.com,DNS:strimzi-cluster-broker-2.tcp.devops.XXXX.XXXX.com"
openssl req -text -noout -verify -in listener.csr | head -1; openssl rsa -check -in listener.key | head -1
verify OK
Certificate Request:
writing RSA key
RSA key ok

Sign with CA and validate

openssl x509 -noout -modulus -in listener.pem | openssl md5;openssl rsa -noout -modulus -in listener.key | openssl md5
(stdin)= 12b02b4de9741c41f718e2cc8157aabe
(stdin)= 12b02b4de9741c41f718e2cc8157aabe

Create secret and apply Kafka CR

kubectl -n kafka create secret generic listener-secret --from-file=listener.crt=listener.pem --from-file=listener.key=listener.key
kubectl apply -f 03-kafka-listener-ephemeral.yaml -n kafka

Validate

openssl s_client -connect strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 -servername strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com -showcerts

Create the topic

kubectl apply -f kafka-topic.yml -n kafka;kubectl get KafkaTopic -n kafka -o wide
kafkatopic.kafka.strimzi.io/test-topic created
NAME                                                                                               CLUSTER           PARTITIONS   REPLICATION FACTOR   READY
consumer-offsets---84e7a678d08f4bd226872e5cdd4eb527fadc1c6a                                        strimzi-cluster   50           3                    True
strimzi-store-topic---effb8e3e057afce1ecf67c3f5d8e4e3ff177fc55                                     strimzi-cluster   1            3                    True
strimzi-topic-operator-kstreams-topic-store-changelog---b75e702040b99be8a9263134de3507fc0cc4017b   strimzi-cluster   1            3                    True
test-topic                                                                                         strimzi-cluster   1            3

Create the trust store for the listener

keytool -import -trustcacerts -alias CARoot -file root.crt -keystore kafka.client.truststore.jks -storepass test1234 -noprompt
keytool -import -trustcacerts -alias CAIntermediate -file inter.crt -keystore kafka.client.truststore.jks -storepass test1234 -noprompt

Produce and consumer the message

kafka-console-producer --broker-list strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 --producer-property security.protocol=SSL --producer-property ssl.truststore.password=test1234 --producer-property ssl.truststore.location=./kafka.client.truststore.jks  --topic test-topic
>Test Message 1
>Test Message 2
>Test Message 3
>^C

kafka-console-consumer --bootstrap-server strimzi-cluster-bootstrap.tcp.devops.XXXX.XXXX.com:443 --consumer-property security.protocol=SSL --consumer-property ssl.truststore.password=test1234 --consumer-property ssl.truststore.location=./kafka.client.truststore.jks --topic test-topic --from-beginning
Test Message 1
Test Message 2
Test Message 3
^C