added support for self-signed certificates

 - now if validateCertificates and respectively a CA is configured they will be used for Kafka messaging. For both consumers and producers.

Signed-off-by: Kalin Kostashki <kalin.kostashki@bosch.io>

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/PropertiesFactory.java

@@ -23,6 +23,7 @@
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.serialization.ByteBufferDeserializer;
 import org.apache.kafka.common.serialization.ByteBufferSerializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -95,6 +96,7 @@ ConsumerSettings<String, ByteBuffer> getConsumerSettings(final boolean dryRun) {
                         .withBootstrapServers(bootstrapServers)
                         .withGroupId(connection.getId().toString())
                         .withClientId(clientId + "-consumer")
+                        .withProperties(getTrustedSelfSignedCertificates())
                         .withProperties(getConsumerSpecificConfigProperties())
                         .withProperties(getSecurityProtocolProperties())
                         .withConnectionChecker(connectionCheckerSettings);
@@ -114,10 +116,19 @@ ProducerSettings<String, ByteBuffer> getProducerSettings() {
         return ProducerSettings.apply(alpakkaConfig, new StringSerializer(), new ByteBufferSerializer())
                 .withBootstrapServers(bootstrapServers)
                 .withProperties(getClientIdProperties())
+                .withProperties(getTrustedSelfSignedCertificates())
                 .withProperties(getProducerSpecificConfigProperties())
                 .withProperties(getSecurityProtocolProperties());
     }
 
+    private Map<String, String> getTrustedSelfSignedCertificates() {
+        if (connection.isValidateCertificates() && connection.getTrustedCertificates().isPresent()) {
+            return Map.of(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM",
+                    SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, connection.getTrustedCertificates().orElse(""));
+        }
+        return Map.of();
+    }
+
     private Map<String, String> getClientIdProperties() {
         return Map.of(CommonClientConfigs.CLIENT_ID_CONFIG, clientId + "-producer");
     }