[STRIMZI] Migrate client from cambria to kafka native

Add call to sdc to get kafka and topic details
Add kafka config to IConfiguration interface

Signed-off-by: efiacor <fiachra.corcoran@est.tech>
Change-Id: Ibec77d1ff1cd25ad4adce133ee81d66e54c7707f
Issue-ID: DMAAP-1745

9 files changed, 241 insertions, 120 deletions

diff --git a/sdc-distribution-ci/etc/asdc-client.jks b/sdc-distribution-ci/etc/sdc-client.jks
index eb0a0d3..eb0a0d3 100644
--- a/sdc-distribution-ci/etc/asdc-client.jks
+++ b/sdc-distribution-ci/etc/sdc-client.jks
diff --git a/sdc-distribution-ci/etc/asdcclientstore.jks b/sdc-distribution-ci/etc/sdcclientstore.jks
index 5dc006d..5dc006d 100644
--- a/sdc-distribution-ci/etc/asdcclientstore.jks
+++ b/sdc-distribution-ci/etc/sdcclientstore.jks
diff --git a/sdc-distribution-ci/pom.xml b/sdc-distribution-ci/pom.xml
index e11d428..707de83 100644
--- a/sdc-distribution-ci/pom.xml
+++ b/sdc-distribution-ci/pom.xml
@@ -7,7 +7,7 @@
   <parent>
     <groupId>org.onap.sdc.sdc-distribution-client</groupId>
     <artifactId>sdc-main-distribution-client</artifactId>
-    <version>1.4.5-SNAPSHOT</version>
+        <version>2.0.0-SNAPSHOT</version>
   </parent>
 
   <artifactId>sdc-distribution-ci</artifactId>
@@ -25,6 +25,13 @@
   </properties>
 
   <dependencies>
+        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>kafka</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
     <dependency>
       <groupId>org.onap.sdc.sdc-distribution-client</groupId>
       <artifactId>sdc-distribution-client</artifactId>
@@ -136,6 +143,11 @@
       </exclusions>
     </dependency>
     <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>2.8.2</version>
+        </dependency>
+        <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
       <version>${mockito-core.version}</version>
@@ -181,6 +193,12 @@
       <version>${httpclient.version}</version>
       <scope>runtime</scope>
     </dependency>
+        <dependency>
+            <groupId>org.junit-pioneer</groupId>
+            <artifactId>junit-pioneer</artifactId>
+            <version>1.4.2</version>
+            <scope>test</scope>
+        </dependency>
   </dependencies>
 
   <build>
@@ -195,7 +213,7 @@
             <manifest>
               <addClasspath>true</addClasspath>
               <classpathPrefix>lib</classpathPrefix>
-              <mainClass>org.onap.test.it.RegisterToAsdcTopicIT</mainClass>
+                            <mainClass>org.onap.test.it.RegisterToSdcTopicIT</mainClass>
             </manifest>
             <manifestEntries>
               <Class-Path>lib/</Class-Path>
diff --git a/sdc-distribution-ci/src/main/java/org/onap/test/core/config/DistributionClientConfig.java b/sdc-distribution-ci/src/main/java/org/onap/test/core/config/DistributionClientConfig.java
index ad18969..5a902c8 100644
--- a/sdc-distribution-ci/src/main/java/org/onap/test/core/config/DistributionClientConfig.java
+++ b/sdc-distribution-ci/src/main/java/org/onap/test/core/config/DistributionClientConfig.java
@@ -26,21 +26,23 @@ import java.util.List;
 
 public class DistributionClientConfig implements IConfiguration {
 
-    public static final String DEFAULT_ASDC_ADDRESS = "localhost:30206";
+    public static final String DEFAULT_SDC_ADDRESS = "localhost:30206";
     public static final String DEFAULT_COMSUMER_ID = "dcae-openapi-manager";
     public static final String DEFAULT_CONSUMER_GROUP = "noapp";
     public static final String DEFAULT_ENVIRONMENT_NAME = "AUTO";
     public static final String DEFAULT_PASSWORD = "Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U";
     public static final int DEFAULT_POLLING_INTERVAL = 20;
     public static final int DEFAULT_POLLING_TIMEOUT = 20;
+    public static final String DEFAULT_STATUS_TOPIC = "STATUS-TOPIC";
+    public static final String DEFAULT_NOTIF_TOPIC = "NOTIF-TOPIC";
     public static final String DEFAULT_USER = "dcae";
-    public static final String DEFAULT_KEY_STORE_PATH = "etc/asdc-client.jks";
+    public static final String DEFAULT_KEY_STORE_PATH = "etc/sdc-client.jks";
     public static final String DEFAULT_KEY_STORE_PASSWORD = "Aa123456";
     public static final boolean DEFAULT_ACTIVATE_SERVER_TLS_AUTH = false;
     public static final boolean DEFAULT_IS_FILTER_IN_EMPTY_RESOURCES = true;
     public static final boolean DEFAULT_USE_HTTPS_WITH_SDC = false;
-    public static final String DEFAULT_MSG_BUS_ADDRESS = "localhost";
-    private String asdcAddress;
+    public static final String DEFAULT_MSG_BUS_ADDRESS = "localhost:9092";
+    private String sdcAddress;
     private String user;
     private String password;
     private int pollingInterval;
@@ -53,38 +55,23 @@ public class DistributionClientConfig implements IConfiguration {
     private String keyStorePassword;
     private boolean activateServerTLSAuth;
     private boolean isFilterInEmptyResources;
-    private boolean useHttpsWithDmaap;
     private boolean useHttpsWithSDC;
     private List<String> msgBusAddress;
+    private String sdcStatusTopicName;
+    private String sdcNotificationTopicName;
+    private String kafkaSecurityProtocolConfig;
+    private String kafkaSaslMechanism;
+    private String kafkaSaslJaasConfig;
     private String httpProxyHost;
     private int httpProxyPort;
     private String httpsProxyHost;
     private int httpsProxyPort;
     private boolean useSystemProxy;
 
-    public DistributionClientConfig(IConfiguration other) {
-        this.asdcAddress = other.getAsdcAddress();
-        this.comsumerID = other.getConsumerID();
-        this.consumerGroup = other.getConsumerGroup();
-        this.environmentName = other.getEnvironmentName();
-        this.password = other.getPassword();
-        this.pollingInterval = other.getPollingInterval();
-        this.pollingTimeout = other.getPollingTimeout();
-        this.relevantArtifactTypes = other.getRelevantArtifactTypes();
-        this.user = other.getUser();
-        this.keyStorePath = other.getKeyStorePath();
-        this.keyStorePassword = other.getKeyStorePassword();
-        this.activateServerTLSAuth = other.activateServerTLSAuth();
-        this.isFilterInEmptyResources = other.isFilterInEmptyResources();
-        this.httpProxyHost = other.getHttpProxyHost();
-        this.httpProxyPort = other.getHttpProxyPort();
-        this.httpsProxyHost = other.getHttpsProxyHost();
-        this.httpsProxyPort = other.getHttpsProxyPort();
-        this.useSystemProxy = other.isUseSystemProxy();
-    }
-
     public DistributionClientConfig() {
-        this.asdcAddress = DEFAULT_ASDC_ADDRESS;
+        this.sdcAddress = DEFAULT_SDC_ADDRESS;
+        this.sdcStatusTopicName = DEFAULT_STATUS_TOPIC;
+        this.sdcNotificationTopicName = DEFAULT_NOTIF_TOPIC;
         this.comsumerID = DEFAULT_COMSUMER_ID;
         this.consumerGroup = DEFAULT_CONSUMER_GROUP;
         this.environmentName = DEFAULT_ENVIRONMENT_NAME;
@@ -99,22 +86,23 @@ public class DistributionClientConfig implements IConfiguration {
         this.activateServerTLSAuth = DEFAULT_ACTIVATE_SERVER_TLS_AUTH;
         this.isFilterInEmptyResources = DEFAULT_IS_FILTER_IN_EMPTY_RESOURCES;
         this.useHttpsWithSDC = DEFAULT_USE_HTTPS_WITH_SDC;
-        msgBusAddress = new ArrayList<>();
-        msgBusAddress.add(DEFAULT_MSG_BUS_ADDRESS);
-        msgBusAddress.add(DEFAULT_MSG_BUS_ADDRESS);
-        msgBusAddress.add(DEFAULT_MSG_BUS_ADDRESS);
+        this.msgBusAddress = new ArrayList<>();
+        this.msgBusAddress.add(DEFAULT_MSG_BUS_ADDRESS);
     }
 
     @Override
-    public String getAsdcAddress() {
-        return asdcAddress;
+    public String getSdcAddress() {
+        return sdcAddress;
     }
 
-    @Override
     public List<String> getMsgBusAddress() {
         return msgBusAddress;
     }
 
+    public void setMsgBusAddress(List<String> msgBusAddress) {
+        this.msgBusAddress = msgBusAddress;
+    }
+
     @Override
     public String getUser() {
         return user;
@@ -173,8 +161,8 @@ public class DistributionClientConfig implements IConfiguration {
         this.comsumerID = comsumerID;
     }
 
-    public void setAsdcAddress(String asdcAddress) {
-        this.asdcAddress = asdcAddress;
+    public void setSdcAddress(String sdcAddress) {
+        this.sdcAddress = sdcAddress;
     }
 
     public void setUser(String user) {
@@ -217,7 +205,7 @@ public class DistributionClientConfig implements IConfiguration {
     public int hashCode() {
         final int prime = 31;
         int result = 1;
-        result = prime * result + ((asdcAddress == null) ? 0 : asdcAddress.hashCode());
+        result = prime * result + ((sdcAddress == null) ? 0 : sdcAddress.hashCode());
         result = prime * result + ((comsumerID == null) ? 0 : comsumerID.hashCode());
         result = prime * result + ((consumerGroup == null) ? 0 : consumerGroup.hashCode());
         result = prime * result + ((environmentName == null) ? 0 : environmentName.hashCode());
@@ -251,11 +239,11 @@ public class DistributionClientConfig implements IConfiguration {
             return false;
         }
         DistributionClientConfig other = (DistributionClientConfig) obj;
-        if (asdcAddress == null) {
-            if (other.asdcAddress != null) {
+        if (sdcAddress == null) {
+            if (other.sdcAddress != null) {
                 return false;
             }
-        } else if (!asdcAddress.equals(other.asdcAddress)) {
+        } else if (!sdcAddress.equals(other.sdcAddress)) {
             return false;
         }
         if (comsumerID == null) {
@@ -322,7 +310,7 @@ public class DistributionClientConfig implements IConfiguration {
 
     @Override
     public String toString() {
-        return "TestConfiguration [asdcAddress=" + asdcAddress + ", user=" + user + ", password=" + password
+        return "TestConfiguration [sdcAddress=" + sdcAddress + ", user=" + user + ", password=" + password
                 + ", pollingInterval=" + pollingInterval + ", pollingTimeout=" + pollingTimeout
                 + ", relevantArtifactTypes=" + relevantArtifactTypes + ", consumerGroup=" + consumerGroup
                 + ", environmentName=" + environmentName + ", comsumerID=" + comsumerID + "]";
@@ -337,11 +325,6 @@ public class DistributionClientConfig implements IConfiguration {
         this.isFilterInEmptyResources = isFilterInEmptyResources;
     }
 
-    @Override
-    public Boolean isUseHttpsWithDmaap() {
-        return this.useHttpsWithDmaap;
-    }
-
     public Boolean isUseHttpsWithSDC() {
         return this.useHttpsWithSDC;
     }
diff --git a/sdc-distribution-ci/src/main/java/org/onap/test/core/service/ClientNotifyCallback.java b/sdc-distribution-ci/src/main/java/org/onap/test/core/service/ClientNotifyCallback.java
index 4dfe388..4cee4cf 100644
--- a/sdc-distribution-ci/src/main/java/org/onap/test/core/service/ClientNotifyCallback.java
+++ b/sdc-distribution-ci/src/main/java/org/onap/test/core/service/ClientNotifyCallback.java
@@ -23,7 +23,7 @@ import org.onap.sdc.api.consumer.IDistributionStatusMessage;
 import org.onap.sdc.api.consumer.INotificationCallback;
 import org.onap.sdc.api.notification.INotificationData;
 import org.onap.sdc.api.notification.IResourceInstance;
-import org.onap.sdc.http.HttpAsdcClient;
+import org.onap.sdc.http.HttpSdcClient;
 import org.onap.sdc.http.SdcConnectorClient;
 import org.onap.sdc.impl.DistributionClientDownloadResultImpl;
 import org.onap.sdc.impl.DistributionClientImpl;
@@ -46,8 +46,8 @@ public class ClientNotifyCallback implements INotificationCallback {
     private final DistributionClientImpl distributionClient;
     private final List<DistributionClientDownloadResultImpl> pulledArtifacts = new ArrayList<>();
     DistributionClientConfig config = new DistributionClientConfig();
-    HttpAsdcClient asdcClient = new HttpAsdcClient(config);
-    SdcConnectorClient sdcConnectorClient = new SdcConnectorClient(config, asdcClient);
+    HttpSdcClient sdcClient = new HttpSdcClient(config);
+    SdcConnectorClient sdcConnectorClient = new SdcConnectorClient(config, sdcClient);
     ArtifactsDownloader artifactsDownloader = new ArtifactsDownloader("/app/path", sdcConnectorClient);
 
     public List<DistributionClientDownloadResultImpl> getPulledArtifacts() {
diff --git a/sdc-distribution-ci/src/main/java/org/onap/test/it/RegisterToAsdcTopicIT.java b/sdc-distribution-ci/src/main/java/org/onap/test/it/RegisterToSdcTopicIT.java
index 58baec7..c89ecd4 100644
--- a/sdc-distribution-ci/src/main/java/org/onap/test/it/RegisterToAsdcTopicIT.java
+++ b/sdc-distribution-ci/src/main/java/org/onap/test/it/RegisterToSdcTopicIT.java
@@ -19,18 +19,15 @@
  */
 package org.onap.test.it;
 
+import java.util.ArrayList;
+import java.util.List;
 import org.onap.sdc.impl.DistributionClientImpl;
 import org.onap.test.core.config.DistributionClientConfig;
 import org.onap.test.core.service.ArtifactsValidator;
 import org.onap.test.core.service.ClientInitializer;
 import org.onap.test.core.service.ClientNotifyCallback;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
 
-public class RegisterToAsdcTopicIT {
+public class RegisterToSdcTopicIT {
 
 
     public static void main(String[] args) {
@@ -40,12 +37,10 @@ public class RegisterToAsdcTopicIT {
         ClientNotifyCallback callback = new ClientNotifyCallback(validators, client);
         ClientInitializer clientInitializer = new ClientInitializer(clientConfig, callback, client);
         clientInitializer.initialize();
-        Runtime.getRuntime().addShutdownHook(new Thread() {
-            public void run() {
-                client.stop();
-                System.out.println("Shutdown Hook is running !");
-            }
-        });
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            client.stop();
+            System.out.println("Shutdown Hook is running !");
+        }));
 
     }
 }
diff --git a/sdc-distribution-ci/src/test/java/org/onap/test/core/service/ClientInitializerTest.java b/sdc-distribution-ci/src/test/java/org/onap/test/core/service/ClientInitializerTest.java
index ba118c0..1abef1f 100644
--- a/sdc-distribution-ci/src/test/java/org/onap/test/core/service/ClientInitializerTest.java
+++ b/sdc-distribution-ci/src/test/java/org/onap/test/core/service/ClientInitializerTest.java
@@ -24,18 +24,35 @@ import static org.awaitility.Awaitility.await;
 import static org.mockito.Mockito.verify;
 
 import java.io.IOException;
-import java.net.URI;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
-import java.nio.file.Paths;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
-import org.awaitility.Durations;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import lombok.SneakyThrows;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.SaslConfigs;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junitpioneer.jupiter.SetEnvironmentVariable;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.Mockito;
@@ -46,84 +63,52 @@ import org.onap.test.core.config.DistributionClientConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.WaitStrategy;
+import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
-
+import org.testcontainers.shaded.org.awaitility.Durations;
+import org.testcontainers.utility.DockerImageName;
 @Testcontainers
 @ExtendWith(MockitoExtension.class)
+@SetEnvironmentVariable(key = "SASL_JAAS_CONFIG", value = "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin-secret';")
+@SetEnvironmentVariable(key = "SECURITY_PROTOCOL", value = "SASL_PLAINTEXT")
+@SetEnvironmentVariable(key = "SASL_MECHANISM", value = "PLAIN")
 class ClientInitializerTest {
 
-    private static final int SUCCESSFUL_STOP_MSG_INDEX = 2;
-    private static final int SUCCESSFUL_UNREGISTER_MSG_INDEX = 3;
-    private static final int SUCCESSFUL_INIT_MSG_INDEX = 0;
-    private static final int SUCCESSFUL_DIST_MSG_INDEX = 3;
     private static final int EXPECTED_HEAT_ARTIFACTS = 4;
+    private static final DistributionClientConfig clientConfig = new DistributionClientConfig();
     private static final Logger testLog = LoggerFactory.getLogger(ClientInitializerTest.class);
     @Container
-    public GenericContainer mockDmaap = new GenericContainer("registry.gitlab.com/orange-opensource/lfn/onap/mock_servers/mock-dmaap:latest")
-        .withNetworkMode("host");
+    CustomKafkaContainer kafka = buildBrokerInstance();
     @Container
-    public GenericContainer mockSdc = new GenericContainer("registry.gitlab.com/orange-opensource/lfn/onap/mock_servers/mock-sdc:latest")
-        .withNetworkMode("host");
-    @Mock
-    private Logger log;
+    public GenericContainer<?> mockSdc =
+        new GenericContainer<>(
+            "registry.gitlab.com/orange-opensource/lfn/onap/mock_servers/mock-sdc:develop")
+            .withExposedPorts(30206);
     @Mock
     private Logger distClientLog;
     private ClientInitializer clientInitializer;
     private ClientNotifyCallback clientNotifyCallback;
 
     @BeforeEach
-    public void initializeClient() {
-        DistributionClientConfig clientConfig = new DistributionClientConfig();
+    public void initializeClient() throws InterruptedException {
+        clientConfig.setSdcAddress(mockSdc.getHost()+":"+mockSdc.getFirstMappedPort());
         List<ArtifactsValidator> validators = new ArrayList<>();
         DistributionClientImpl client = new DistributionClientImpl(distClientLog);
         clientNotifyCallback = new ClientNotifyCallback(validators, client);
         clientInitializer = new ClientInitializer(clientConfig, clientNotifyCallback, client);
-    }
-
-    @Test
-    void shouldRegisterToDmaapAfterClientInitialization() {
-        //given
-        final ArgumentCaptor<String> exceptionCaptor = ArgumentCaptor.forClass(String.class);
-        //when
-        clientInitializer.log = log;
-        clientInitializer.initialize();
-        verify(log, Mockito.atLeastOnce()).info(exceptionCaptor.capture());
-        List<String> allValues = exceptionCaptor.getAllValues();
-        //then
-        assertThat(allValues.get(SUCCESSFUL_INIT_MSG_INDEX)).isEqualTo("distribution client initialized successfuly");
-        assertThat(allValues.get(SUCCESSFUL_DIST_MSG_INDEX)).isEqualTo("distribution client started successfuly");
-    }
-
-    @Test
-    void shouldUnregisterAndStopClient() {
-        //given
-        final ArgumentCaptor<String> exceptionCaptor = ArgumentCaptor.forClass(String.class);
-        //when
         clientInitializer.initialize();
-        clientInitializer.stop();
-        verify(distClientLog, Mockito.atLeastOnce()).info(exceptionCaptor.capture());
-        List<String> allValues = exceptionCaptor.getAllValues();
-        //then
-        assertThat(allValues.get(SUCCESSFUL_STOP_MSG_INDEX)).isEqualTo("stop DistributionClient");
-        assertThat(allValues.get(SUCCESSFUL_UNREGISTER_MSG_INDEX)).isEqualTo("client unregistered from topics successfully");
+        Thread.sleep(1000);
+        setUpTopicsAndSendData();
     }
 
     @Test
-    void shouldDownloadArtifactsWithArtifactTypeHeat() throws IOException, InterruptedException {
-
-        //given
-        HttpRequest request = HttpRequest.newBuilder()
-            .uri(URI.create("http://localhost:3904/events/testName/add"))
-            .POST(HttpRequest.BodyPublishers.ofFile(Paths.get("src/test/resources/artifacts.json")))
-            .build();
-        HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofString());
-        //when
-        clientInitializer.initialize();
+    void shouldDownloadArtifactsWithArtifactTypeHeat() {
         waitForArtifacts();
         List<DistributionClientDownloadResultImpl> calls = clientNotifyCallback.getPulledArtifacts();
-        //then
         Assertions.assertEquals(EXPECTED_HEAT_ARTIFACTS, calls.size());
+        clientInitializer.stop();
     }
 
     private void waitForArtifacts() {
@@ -132,4 +117,36 @@ class ClientInitializerTest {
             .atMost(Durations.ONE_MINUTE)
             .until(() -> !clientNotifyCallback.getPulledArtifacts().isEmpty());
     }
+
+    private CustomKafkaContainer buildBrokerInstance() {
+        final Map<String, String> env = new LinkedHashMap<>();
+        env.put("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:SASL_PLAINTEXT");
+        env.put("KAFKA_SASL_ENABLED_MECHANISMS", "PLAIN");
+        env.put("KAFKA_LISTENER_NAME_PLAINTEXT_PLAIN_SASL_JAAS_CONFIG", "org.apache.kafka.common.security.plain.PlainLoginModule required " +
+            "username=\"admin\" " +
+            "password=\"admin-secret\" " +
+            "user_admin=\"admin-secret\" " +
+            "user_producer=\"producer-secret\" " +
+            "user_consumer=\"consumer-secret\";");
+
+        return new CustomKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"))
+            .withEmbeddedZookeeper()
+            .withStartupAttempts(1)
+            .withEnv(env)
+            .withFixedExposedPort(43219, 9093);
+    }
+
+    @SneakyThrows
+    private void setUpTopicsAndSendData()  {
+        Properties props = new Properties();
+        props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='admin-secret';");
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,  "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+        String content = Files.readString(Path.of("src/test/resources/artifacts.json"));
+        producer.send(new ProducerRecord<>("SDC-DIST-NOTIF-TOPIC", "testcontainers", content)).get();
+    }
 }
diff --git a/sdc-distribution-ci/src/test/java/org/onap/test/core/service/CustomKafkaContainer.java b/sdc-distribution-ci/src/test/java/org/onap/test/core/service/CustomKafkaContainer.java
new file mode 100644
index 0000000..e2eabc1
--- /dev/null
+++ b/sdc-distribution-ci/src/test/java/org/onap/test/core/service/CustomKafkaContainer.java
@@ -0,0 +1,94 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * sdc-distribution-client
+ * ================================================================================
+ * Copyright (C) 2022 Nordix Foundation. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.test.core.service;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import lombok.SneakyThrows;
+import org.testcontainers.containers.FixedHostPortGenericContainer;
+import org.testcontainers.utility.DockerImageName;
+
+public class CustomKafkaContainer extends FixedHostPortGenericContainer<CustomKafkaContainer> {
+    protected String externalZookeeperConnect;
+
+    public CustomKafkaContainer(DockerImageName dockerImageName) {
+        super(String.valueOf(dockerImageName));
+        this.externalZookeeperConnect = null;
+        this.withExposedPorts(9093);
+        this.withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092");
+        this.withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "BROKER:PLAINTEXT,PLAINTEXT:PLAINTEXT");
+        this.withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "BROKER");
+        this.withEnv("KAFKA_BROKER_ID", "1");
+        this.withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1");
+        this.withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1");
+        this.withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1");
+        this.withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
+        this.withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", "9223372036854775807");
+        this.withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "0");
+    }
+
+    public CustomKafkaContainer withEmbeddedZookeeper() {
+        this.externalZookeeperConnect = null;
+        return this.self();
+    }
+
+    public String getBootstrapServers() {
+        return String.format("PLAINTEXT://%s:%s", this.getHost(), this.getMappedPort(9093));
+    }
+
+    protected void configure() {
+        this.withEnv("KAFKA_ADVERTISED_LISTENERS", String.format("BROKER://%s:9092", this.getNetwork() != null ? this.getNetworkAliases().get(0) : "localhost"));
+        String command = "";
+        if (this.externalZookeeperConnect != null) {
+            this.withEnv("KAFKA_ZOOKEEPER_CONNECT", this.externalZookeeperConnect);
+        } else {
+            this.addExposedPort(2181);
+            this.withEnv("KAFKA_ZOOKEEPER_CONNECT", "localhost:2181");
+            command = command + "echo 'clientPort=2181' > zookeeper.properties\n";
+            command = command + "echo 'dataDir=/var/lib/zookeeper/data' >> zookeeper.properties\n";
+            command = command + "echo 'dataLogDir=/var/lib/zookeeper/log' >> zookeeper.properties\n";
+            command = command + "zookeeper-server-start zookeeper.properties &\n";
+        }
+
+        command = command + "echo '' > /etc/confluent/docker/ensure \n";
+        command = command + "/etc/confluent/docker/run \n";
+        this.withCommand("sh", "-c", command);
+    }
+
+    @SneakyThrows
+    protected void containerIsStarted(InspectContainerResponse containerInfo) {
+        try {
+            String brokerAdvertisedListener = this.brokerAdvertisedListener(containerInfo);
+            ExecResult result = this.execInContainer("kafka-configs", "--alter", "--bootstrap-server",
+                brokerAdvertisedListener, "--entity-type", "brokers", "--entity-name",
+                this.getEnvMap().get("KAFKA_BROKER_ID"), "--add-config",
+                "advertised.listeners=[" + String.join(",", this.getBootstrapServers(), brokerAdvertisedListener) + "]");
+            if (result.getExitCode() != 0) {
+                throw new IllegalStateException(result.toString());
+            }
+        } catch (Throwable var4) {
+            throw var4;
+        }
+    }
+
+    protected String brokerAdvertisedListener(InspectContainerResponse containerInfo) {
+        return String.format("BROKER://%s:%s", containerInfo.getConfig().getHostName(), "9092");
+    }
+}
diff --git a/sdc-distribution-ci/src/test/resources/logback-test.xml b/sdc-distribution-ci/src/test/resources/logback-test.xml
new file mode 100644
index 0000000..78c5b05
--- /dev/null
+++ b/sdc-distribution-ci/src/test/resources/logback-test.xml
@@ -0,0 +1,14 @@
+<configuration>
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <root level="INFO">
+    <appender-ref ref="STDOUT"/>
+  </root>
+
+  <logger name="org.testcontainers" level="INFO"/>
+  <logger name="com.github.dockerjava" level="WARN"/>
+</configuration>
\ No newline at end of file