SSL kafka connection configuration

[BHUVANESHWARI967]

Hello Everyone,

I am currently trying to integrate Kafka with Ditto and I need some assistance in setting up the connection configuration JSON.I have configured SSL to secure the connections between my clients and broker.

Security and Authentication: SSL/TLS (configured with keystores and truststores).

I am unsure about the fields required, especially in relation to Kafka specifics and Ditto integration.

Could anyone guide me on the correct and necessary fields for this JSON configuration when connecting Kafka with Ditto?

[thjaeckle]

@BHUVANESHWARI967 it is unclear to me of you configure SSL/TLS for verifying a server certificate (documented at Certificates for Transport Layer Security - Verify server certificate) or for client authentication (documented at Certificates for Transport Layer Security - Authenticate by client certificate) at the Kafka broker.

If it is the first one, the documentation states in which format the ca has to be provided:

ca: A string of trusted certificates as PEM-encoded DER. Concatenate multiple certificates as strings to trust all of them. Omit to trust popular certificate authorities.

Maybe you are missing the mandatory line breaks in PEM or DER format?

If it is the second one (client certificate authentication), the documentation clearly states:

Client-certificate authentication is available for MQTT connections and HTTP connections only.

[BHUVANESHWARI967]

Hi @thjaeckle ,

Thanks for replying. I am getting this error even after encoding the cert to required format.

{
"status": 400,
"error": "connectivity:connection.configuration.invalid",
"message": "/ca: bad format. Expect PEM-encoded DER data specified by RFC-7468 starting with '-----BEGIN CERTIFICATE-----'",
"description": "Make sure all required properties are set and valid in the configuration."
}

(Edited)
My connection json is,

{
"name": "Secure Kafka",
"connectionType": "kafka",
"connectionStatus": "open",
"uri": "ssl://32.4.564.23:9093",
"sources": [
{
"addresses": ["theTopic"],
"consumerCount": 1,
"qos": 0,
"authorizationContext": ["nginx:ditto"],
"headerMapping": {},
"payloadMapping": ["Ditto"],
"replyTarget": {
"address": "theTopic-Reply",
"headerMapping": {},
"expectedResponseTypes": ["response", "error"],
"enabled": true
}
}
],
"targets": [
{
"address": "theTopic-Target",
"topics": ["//things/twin/events"],
"qos": 0,
"authorizationContext": ["nginx:ditto"],
"headerMapping": {}
}
],
"clientCount": 1,
"failoverEnabled": true,
"validateCertificates": true,
"ca": "-----BEGIN CERTIFICATE-----{certdata}-----END CERTIFICATE-----",
"processorPoolSize": 1,
"specificConfig": {
"saslMechanism": "plain",
"bootstrapServers": "32.4.564.23:9093"
},
"mappingDefinitions": {
"ditto": {
"mappingEngine": "",
"options": {
"incomingScript": "",
"outgoingScript": ""
}
}
},
"tags": []
}

{
"status": 504,
"error": "connectivity:connection.failed",
"message": "The Connection with ID '1d56cf82-6e59-4e78-b7fe-05dd0a35f5f0' failed to connect.",
"description": "Cause: Unexpected handshake message: server_hello"
}

connection-log.txt

[alstanchev]

When configuring the certificate in the kafka connection json did you add \n at each line of the certificate. To put a multiline string in the json and for it to be properly interpreted later you should define the new lines. So the string of the certificate should be one long line with \n where a new line is expected.
For example in the cloud2edge packages repo this is done by a sed command in the post-instal.sh script here

[BHUVANESHWARI967]

Hi @alstanchev,

Thanks for the reply. I have included the \n at each line of the certificate. But still getting the SSLHandshakeException. I have created a issue https://github.com/eclipse-ditto/ditto/issues/1935.

Can you tell me if there's anything to be added in the deployment yaml of ditto for making Tls kafka connection? Since I am getting this error in connection,

'{
"status": 504,
"error": "connectivity:connection.failed",
"message": "The Connection with ID '4c4787b3-a26b-40d8-997c-fe07e6e42aad' failed to connect.",
"description": "Cause: unable to find valid certification path to requested target"
}'

[alstanchev]

To the deployment no, all the necessery configuration should be in the connection. The trusted certificate is the "ca" field so no need to add it to the JVM on start. Maybe add -Djavax.net.debug=SSL to see where the handshake breaks.

[BHUVANESHWARI967]

Hi @alstanchev,
-Djavax.net.debug=SSL in connectivity deployment yaml as a env.

The connection log after this was,
2024-05-09 06:50:54,607 ERROR [][] o.a.p.k.i.KafkaConsumerActor pekko://ditto-cluster@10.42.0.223:2551/system/kafka-consumer-67 - [1d455] Exception when polling from consumer, stopping actor: org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
org.apache.kafka.common.errors.SslAuthenticationException: SSL handshake failed
Caused by: javax.net.ssl.SSLProtocolException: Unexpected handshake message: server_hello
at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:129)
at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:117)
at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:365)
at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:321)
at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:312)
at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:474)
at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1277)
at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1264)
at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1209)
at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:435)
at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:523)
at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:308)
at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:571)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:280)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:251)
at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:571)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:280)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1323)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1254)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1227)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.poll(KafkaConsumerActor.scala:516)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.commitAndPoll(KafkaConsumerActor.scala:502)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.org$apache$pekko$kafka$internal$KafkaConsumerActor$$receivePoll(KafkaConsumerActor.scala:486)
at org.apache.pekko.kafka.internal.KafkaConsumerActor$$anonfun$regularReceive$1.applyOrElse(KafkaConsumerActor.scala:264)
at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)
at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.org$apache$pekko$actor$Timers$$super$aroundReceive(KafkaConsumerActor.scala:184)
at org.apache.pekko.actor.Timers.aroundReceive(Timers.scala:62)
at org.apache.pekko.actor.Timers.aroundReceive$(Timers.scala:51)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.aroundReceive(KafkaConsumerActor.scala:184)
at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:251)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1323)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1254)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1227)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.poll(KafkaConsumerActor.scala:516)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.commitAndPoll(KafkaConsumerActor.scala:502)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.org$apache$pekko$kafka$internal$KafkaConsumerActor$$receivePoll(KafkaConsumerActor.scala:486)
at org.apache.pekko.kafka.internal.KafkaConsumerActor$$anonfun$regularReceive$1.applyOrElse(KafkaConsumerActor.scala:264)
at org.apache.pekko.actor.Actor.aroundReceive(Actor.scala:547)
at org.apache.pekko.actor.Actor.aroundReceive$(Actor.scala:545)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.org$apache$pekko$actor$Timers$$super$aroundReceive(KafkaConsumerActor.scala:184)
at org.apache.pekko.actor.Timers.aroundReceive(Timers.scala:62)
at org.apache.pekko.actor.Timers.aroundReceive$(Timers.scala:51)
at org.apache.pekko.kafka.internal.KafkaConsumerActor.aroundReceive(KafkaConsumerActor.scala:184)
at org.apache.pekko.actor.ActorCell.receiveMessage(ActorCell.scala:590)
at org.apache.pekko.actor.ActorCell.invoke(ActorCell.scala:557)
at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)
at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)
at org.apache.pekko.dispatch.Mailbox.processMailbox(Mailbox.scala:280)
at org.apache.pekko.dispatch.Mailbox.run(Mailbox.scala:241)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:840)
2024-05-09 06:50:54,607 INFO [][] o.e.d.c.s.m.k.KafkaClientActor pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1 - CONNECTING failed: <ImmutableConnectionFailure [origin=Actor[pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1/kafkaConsumer-0-kafkatls-01#-1162384521], cause=: SSL handshake failed. Cause: : Unexpected handshake message: server_hello., description=Unexpected consumer failure., time=2024-05-09T04:50:54.607325582Z, connectivityStatus=null]>
2024-05-09 06:50:54,607 WARN [d9d98f26-32f3-4d41-b7dc-4f3fa4b7e45a][] o.e.d.c.s.m.p.ConnectionPersistenceActor pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa - Operation on connection <4c4787b3-a26b-40d8-997c-fe07e6e42aad> failed due to ConnectionFailedException: The Connection with ID '4c4787b3-a26b-40d8-997c-fe07e6e42aad' failed to connect..
2024-05-09 06:50:54,608 WARN [][] o.e.d.c.s.m.k.KafkaConsumerActor pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1/kafkaConsumer-0-kafkatls-01 - Error when shutting down the kafka consumer stream for connection with ID <4c4787b3-a26b-40d8-997c-fe07e6e42aad>
2024-05-09 06:50:54,608 INFO [][] o.e.d.c.s.m.k.KafkaClientActor pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1 - Connection failed: ImmutableConnectionFailure [origin=Actor[pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1/kafkaConsumer-0-kafkatls-01#-1162384521], cause=: SSL handshake failed. Cause: : Unexpected handshake message: server_hello., description=Unexpected consumer failure., time=2024-05-09T04:50:54.607325582Z, connectivityStatus=null]. Reconnect after: PT5S. Resolved status: misconfigured. Going to 'CONNECTING'
2024-05-09 06:50:54,608 WARN [][] o.e.d.c.s.m.k.KafkaConsumerActor pekko://ditto-cluster/system/sharding/connection/1/4c4787b3-a26b-40d8-997c-fe07e6e42aad/pa/$f/c1/kafkaConsumer-0-kafkatls-01 - Error when shutting down the kafka consumer stream for connection with ID <4c4787b3-a26b-40d8-997c-fe07e6e42aad>
2024-05-09 06:50:55,569 INFO [persistence-ping-actor-triggered:8b3f3f51-2309-47d6-9ade-cbf222901f65][] o.e.d.c.s.m.p.ConnectionPersistenceActor pekko://ditto-cluster/system/sharding/connection/12/8b3f3f51-2309-47d6-9ade-cbf222901f65/pa - Calculated live ConnectionStatus: <RetrieveConnectionStatusResponse [connectionId=8b3f3f51-2309-47d6-9ade-cbf222901f65, jsonObject={"type":"connectivity.responses:retrieveConnectionStatus","status":200,"connectionId":"8b3f3f51-2309-47d6-9ade-cbf222901f65","connectionStatus":"open","liveStatus":"open","recoveryStatus":"succeeded","connectedSince":"2024-05-09T04:28:31.574528312Z","clientStatus":[{"type":"client","client":"ditto-connectivity-64b6d689d7-xqv9j","status":"open","recoveryStatus":"succeeded","statusDetails":"CONNECTED","inStateSince":"2024-05-09T04:28:31.574528312Z"}],"sourceStatus":[{"type":"source","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"OTAtopic-7","status":"open","statusDetails":"Consumer started.","inStateSince":"2024-05-09T04:28:25.214174200Z"}],"targetStatus":[{"type":"target","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"ddl.campaign.nirman","status":"open","statusDetails":"Producer started.","inStateSince":"2024-05-09T04:28:25.200060676Z"}]}]>
2024-05-09 06:50:56,570 INFO [persistence-ping-actor-triggered:4dc35934-6da7-4ab4-b5f0-7a78e18a1de4][] o.e.d.c.s.m.p.ConnectionPersistenceActor pekko://ditto-cluster/system/sharding/connection/19/4dc35934-6da7-4ab4-b5f0-7a78e18a1de4/pa - Calculated live ConnectionStatus: <RetrieveConnectionStatusResponse [connectionId=4dc35934-6da7-4ab4-b5f0-7a78e18a1de4, jsonObject={"type":"connectivity.responses:retrieveConnectionStatus","status":200,"connectionId":"4dc35934-6da7-4ab4-b5f0-7a78e18a1de4","connectionStatus":"open","liveStatus":"open","recoveryStatus":"succeeded","connectedSince":"2024-05-09T04:28:40.571788231Z","clientStatus":[{"type":"client","client":"ditto-connectivity-64b6d689d7-xqv9j","status":"open","recoveryStatus":"succeeded","statusDetails":"CONNECTED","inStateSince":"2024-05-09T04:28:40.571788231Z"}],"sourceStatus":[{"type":"source","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"kafkaLexi","status":"open","statusDetails":"Consumer started.","inStateSince":"2024-05-09T04:28:33.775694189Z"}],"targetStatus":[{"type":"target","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"kafkaLexi-Target","status":"open","statusDetails":"Producer started.","inStateSince":"2024-05-09T04:28:33.775329894Z"}]}]>
2024-05-09 06:50:57,571 INFO [persistence-ping-actor-triggered:ee5727ed-e458-4fd9-807c-b9a2bea8756d][] o.e.d.c.s.m.p.ConnectionPersistenceActor pekko://ditto-cluster/system/sharding/connection/1/ee5727ed-e458-4fd9-807c-b9a2bea8756d/pa - Calculated live ConnectionStatus: <RetrieveConnectionStatusResponse [connectionId=ee5727ed-e458-4fd9-807c-b9a2bea8756d, jsonObject={"type":"connectivity.responses:retrieveConnectionStatus","status":200,"connectionId":"ee5727ed-e458-4fd9-807c-b9a2bea8756d","connectionStatus":"open","liveStatus":"open","recoveryStatus":"succeeded","connectedSince":"2024-05-09T04:28:40.571662469Z","clientStatus":[{"type":"client","client":"ditto-connectivity-64b6d689d7-xqv9j","status":"open","recoveryStatus":"succeeded","statusDetails":"CONNECTED","inStateSince":"2024-05-09T04:28:40.571662469Z"}],"sourceStatus":[{"type":"source","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"OTAtopic-7","status":"open","statusDetails":"Consumer started.","inStateSince":"2024-05-09T04:28:33.767930955Z"}],"targetStatus":[{"type":"target","client":"ditto-connectivity-64b6d689d7-xqv9j","address":"ddl.campaign.nirman","status":"open","statusDetails":"Producer started.","inStateSince":"2024-05-09T04:28:33.766950737Z"}]}]>