MQTT source connector throwing the exception

[pious001]

Hello,

Has anyone come across or can shed some light on the issue I am facing? I recently downloaded the kafka-connect-mqtt-assembly-11.3.0.jar to use to connect to a remote MQTT server. Everything seems to be fine except that a certificate path validation error.

I am using the below connector file. Can anyone help with the parameters and the format for supplying the certificates. The handshake between kafka-connect server and mqtt server will be using 1-way SSL.

{
"name": "SourceConnector_MQTT",
"config": {
"connector.class":"io.lenses.streamreactor.connect.mqtt.source.MqttSourceConnector",
"connect.mqtt.kcql":"INSERT INTO kafka-topic SELECT * FROM mqtt/#",
"connect.mqtt.client.id":"3f9b8c2e-7d4a-4f1b-9a6e-2c5d8f7e1b3a",
"connect.mqtt.username":"testuser",
"connect.mqtt.password":"testuser",
"connect.mqtt.ssl.ca.cert":"/kafka/cert/ca.cer",
"connect.mqtt.ssl.cert":"/kafka/cert/client.cer",
"connect.mqtt.ssl.key":"/kafka/cert/client.key",
"connect.mqtt.service.quality":"1",
"connect.mqtt.hosts":"ssl://hostname:port",
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter":"org.apache.kafka.connect.storage.StringConverter"
}
}

exception
"trace":"MqttException (0) - javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
\tat org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
\tat org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:738)
\tat java.base/java.lang.Thread.run(Thread.java:829)
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
\tat java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
\tat java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:366)
\tat java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:309)
\tat java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:304)
\tat java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1357)
\tat java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1232)
\tat java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175)
\tat java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
\tat java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
\tat java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:421)
\tat java.base/sun.security.ssl.TransportContext.dispatch(TransportContext.java:189)
\tat java.base/sun.security.ssl.SSLTransport.decode(SSLTransport.java:172)
\tat java.base/sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1511)
\tat java.base/sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1421)
\tat java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:456)
\tat java.base/sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:427)
\tat org.eclipse.paho.client.mqttv3.internal.SSLNetworkModule.start(SSLNetworkModule