[KAFKA] Allow kafka params to be passed as config

Allow topic names to be passed
Add new api endpoint to retrieve the kafka and topic info

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

12 files changed, 585 insertions, 272 deletions

diff --git a/catalog-be/src/main/docker/backend/chef-repo/cookbooks/sdc-catalog-be/templates/default/BE-distribution-engine-configuration.yaml.erb b/catalog-be/src/main/docker/backend/chef-repo/cookbooks/sdc-catalog-be/templates/default/BE-distribution-engine-configuration.yaml.erb
index 7eeb7a8ff8..0989def1f9 100644
--- a/catalog-be/src/main/docker/backend/chef-repo/cookbooks/sdc-catalog-be/templates/default/BE-distribution-engine-configuration.yaml.erb
+++ b/catalog-be/src/main/docker/backend/chef-repo/cookbooks/sdc-catalog-be/templates/default/BE-distribution-engine-configuration.yaml.erb
@@ -6,8 +6,19 @@ uebServers:
 uebPublicKey: <%= node['UEB']['PublicKey'] %>
 uebSecretKey: <%= node['UEB']['SecretKey'] %>
 
+<% if node.exist?('DistributionTopics','notificationTopicName') -%>
+distributionNotifTopicName: <%= node['DistributionTopics']['notificationTopicName'] %>
+<% else %>
 distributionNotifTopicName: SDC-DISTR-NOTIF-TOPIC
+<% end -%>
+
+<% if node.exist?('DistributionTopics','statusTopicName') -%>
+distributionStatusTopicName: <%= node['DistributionTopics']['statusTopicName'] %>
+<% else %>
 distributionStatusTopicName: SDC-DISTR-STATUS-TOPIC
+<% end -%>
+
+kafkaBootStrapServers: <%= node['Kafka']['bootstrap'] %>
 
 initRetryIntervalSec: 5
 initMaxIntervalSec: 60
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/DistributionBusinessLogic.java b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/DistributionBusinessLogic.java
index d293e9b615..7d9c7357c2 100644
--- a/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/DistributionBusinessLogic.java
+++ b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/DistributionBusinessLogic.java
@@ -42,6 +42,7 @@ import org.openecomp.sdc.be.config.ConfigurationManager;
 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
 import org.openecomp.sdc.be.dao.api.ActionStatus;
 import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus;
+import org.openecomp.sdc.be.distribution.api.client.KafkaDataResponse;
 import org.openecomp.sdc.be.distribution.api.client.RegistrationRequest;
 import org.openecomp.sdc.be.distribution.api.client.ServerListResponse;
 import org.openecomp.sdc.be.distribution.api.client.TopicRegistrationResponse;
@@ -102,6 +103,26 @@ public class DistributionBusinessLogic {
         }
     }
 
+    public Either<KafkaDataResponse, ResponseFormat> getKafkaData() {
+        DistributionEngineConfiguration distributionEngineConfiguration = ConfigurationManager.getConfigurationManager()
+            .getDistributionEngineConfiguration();
+        String bootStrapServers = distributionEngineConfiguration.getKafkaBootStrapServers();
+        if (bootStrapServers != null) {
+            String statusTopicName = DistributionEngineInitTask
+                .buildTopicName(distributionEngineConfiguration.getDistributionStatusTopicName(), distributionEngineConfiguration.getEnvironments().get(0));
+            String notificationTopicName = DistributionEngineInitTask
+                .buildTopicName(distributionEngineConfiguration.getDistributionNotifTopicName(), distributionEngineConfiguration.getEnvironments().get(0));
+            KafkaDataResponse kafkaDataResponse = new KafkaDataResponse();
+            kafkaDataResponse.setKafkaBootStrapServer(bootStrapServers);
+            kafkaDataResponse.setDistrStatusTopicName(statusTopicName);
+            kafkaDataResponse.setDistrNotificationTopicName(notificationTopicName);
+            return Either.left(kafkaDataResponse);
+        } else {
+            ResponseFormat errorResponseWrapper = getResponseFormatManager().getResponseFormat(ActionStatus.GENERAL_ERROR);
+            return Either.right(errorResponseWrapper);
+        }
+    }
+
     public void handleRegistration(Wrapper<Response> responseWrapper, RegistrationRequest registrationRequest, AuditHandler auditHandler) {
         CambriaErrorResponse registerResponse = null;
         try {
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/api/client/KafkaDataResponse.java b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/api/client/KafkaDataResponse.java
new file mode 100644
index 0000000000..db61e884c5
--- /dev/null
+++ b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/api/client/KafkaDataResponse.java
@@ -0,0 +1,35 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * SDC
+ * ================================================================================
+ * 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.openecomp.sdc.be.distribution.api.client;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@Getter
+@Setter
+@NoArgsConstructor
+public class KafkaDataResponse {
+
+    private String kafkaBootStrapServer;
+    private String distrNotificationTopicName;
+    private String distrStatusTopicName;
+}
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/servlet/DistributionServlet.java b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/servlet/DistributionServlet.java
index ee28d948e0..fb8a34830d 100644
--- a/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/servlet/DistributionServlet.java
+++ b/catalog-be/src/main/java/org/openecomp/sdc/be/distribution/servlet/DistributionServlet.java
@@ -49,6 +49,7 @@ import org.openecomp.sdc.be.config.BeEcompErrorManager;
 import org.openecomp.sdc.be.dao.api.ActionStatus;
 import org.openecomp.sdc.be.distribution.AuditHandler;
 import org.openecomp.sdc.be.distribution.DistributionBusinessLogic;
+import org.openecomp.sdc.be.distribution.api.client.KafkaDataResponse;
 import org.openecomp.sdc.be.distribution.api.client.RegistrationRequest;
 import org.openecomp.sdc.be.distribution.api.client.ServerListResponse;
 import org.openecomp.sdc.be.distribution.api.client.TopicRegistrationResponse;
@@ -150,6 +151,58 @@ public class DistributionServlet extends BeGenericServlet {
     }
 
     /**
+     * @param requestId UUID to track the incoming request
+     * @param instanceId UUID to identify the requesting instance
+     * @param accept Determines the format of the body of the response
+     * @param authorization Username and password auth towards SDC
+     * @return KafkaDataResponse (Kafka bootstrap server and topic list to be used by clients)
+     */
+    @GET
+    @Path("/distributionKafkaData")
+    @Consumes(MediaType.APPLICATION_JSON)
+    @Produces(MediaType.APPLICATION_JSON)
+    @Operation(description = "Kafka data", method = "GET", summary = "return the kafka cluster and topic list", responses = {
+        @ApiResponse(responseCode = "200", description = "ECOMP component is authenticated and kafka endpoint and topic list is returned", content = @Content(array = @ArraySchema(schema = @Schema(implementation = KafkaDataResponse.class)))),
+        @ApiResponse(responseCode = "400", description = "Missing 'X-ECOMP-InstanceID'  HTTP header - POL5001"),
+        @ApiResponse(responseCode = "401", description = "ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its credentials  for  Basic Authentication - POL5002"),
+        @ApiResponse(responseCode = "403", description = "ECOMP component is not authorized - POL5003"),
+        @ApiResponse(responseCode = "405", description = "Method  Not Allowed: Invalid HTTP method type used ( PUT,DELETE,POST will be rejected) - POL4050"),
+        @ApiResponse(responseCode = "500", description = "The GET request failed either due to internal SDC problem or Cambria Service failure. ECOMP Component should continue the attempts to get the needed information - POL5000")})
+    public Response getKafkaData(
+        @Parameter(description = "X-ECOMP-RequestID header", required = false) @HeaderParam(value = Constants.X_ECOMP_REQUEST_ID_HEADER) String requestId,
+        @Parameter(description = "X-ECOMP-InstanceID header", required = true) @HeaderParam(value = Constants.X_ECOMP_INSTANCE_ID_HEADER) String instanceId,
+        @Parameter(description = "Determines the format of the body of the response", required = false) @HeaderParam(value = Constants.ACCEPT_HEADER) String accept,
+        @Parameter(description = "The username and password", required = true) @HeaderParam(value = Constants.AUTHORIZATION_HEADER) String authorization) {
+        String url = request.getMethod() + " " + request.getRequestURI();
+        log.debug(START_HANDLE_REQUEST_OF, url);
+        ResponseFormat responseFormat;
+        if (instanceId == null) {
+            responseFormat = getComponentsUtils().getResponseFormat(ActionStatus.MISSING_X_ECOMP_INSTANCE_ID);
+            getComponentsUtils().auditGetUebCluster(null, responseFormat.getStatus().toString(), responseFormat.getFormattedMessage());
+            return buildErrorResponse(responseFormat);
+        }
+        try {
+            Response response;
+            Either<KafkaDataResponse, ResponseFormat> actionResponse = distributionLogic.getKafkaData();
+            if (actionResponse.isRight()) {
+                responseFormat = actionResponse.right().value();
+                response = buildErrorResponse(responseFormat);
+            } else {
+                responseFormat = getComponentsUtils().getResponseFormat(ActionStatus.OK);
+                response = buildOkResponse(responseFormat, actionResponse.left().value());
+            }
+            getComponentsUtils().auditGetUebCluster(instanceId, responseFormat.getStatus().toString(), responseFormat.getFormattedMessage());
+            return response;
+        } catch (Exception e) {
+            BeEcompErrorManager.getInstance().logBeRestApiGeneralError("failed to get kafka cluster and topic list from configuration");
+            log.debug("failed to get kafka cluster and topic list from configuration", e);
+            responseFormat = getComponentsUtils().getResponseFormat(ActionStatus.GENERAL_ERROR);
+            getComponentsUtils().auditGetUebCluster(instanceId, responseFormat.getStatus().toString(), responseFormat.getFormattedMessage());
+            return buildErrorResponse(responseFormat);
+        }
+    }
+
+    /**
      * @param requestId
      * @param instanceId
      * @param accept
diff --git a/catalog-be/src/main/resources/config/distribution-engine-configuration.yaml b/catalog-be/src/main/resources/config/distribution-engine-configuration.yaml
index 1f8d59bfcb..cc7a38d498 100644
--- a/catalog-be/src/main/resources/config/distribution-engine-configuration.yaml
+++ b/catalog-be/src/main/resources/config/distribution-engine-configuration.yaml
@@ -12,6 +12,8 @@ uebSecretKey: 4ZRPzNJfEUK0sSNBvccd2m7X
 distributionNotifTopicName: ASDC-DISTR-NOTIF-TOPIC
 distributionStatusTopicName: ASDC-DISTR-STATUS-TOPIC
 
+kafkaBootStrapServers: kafka-bootstrap:9092
+
 initRetryIntervalSec: 5
 initMaxIntervalSec: 60
 
diff --git a/catalog-be/src/test/resources/config/catalog-be/distribution-engine-configuration.yaml b/catalog-be/src/test/resources/config/catalog-be/distribution-engine-configuration.yaml
index e03b33a87e..8fe5e29f6f 100644
--- a/catalog-be/src/test/resources/config/catalog-be/distribution-engine-configuration.yaml
+++ b/catalog-be/src/test/resources/config/catalog-be/distribution-engine-configuration.yaml
@@ -9,6 +9,8 @@ uebSecretKey: secret
 distributionNotifTopicName: ASDC-DISTR-NOTIF-TOPIC
 distributionStatusTopicName: ASDC-DISTR-STATUS-TOPIC
 
+kafkaBootStrapServers: kafka-bootstrap:9092
+
 initRetryIntervalSec: 5
 initMaxIntervalSec: 60
 
diff --git a/common-app-api/src/main/java/org/openecomp/sdc/be/config/DistributionEngineConfiguration.java b/common-app-api/src/main/java/org/openecomp/sdc/be/config/DistributionEngineConfiguration.java
index 834d7cb579..205588adab 100644
--- a/common-app-api/src/main/java/org/openecomp/sdc/be/config/DistributionEngineConfiguration.java
+++ b/common-app-api/src/main/java/org/openecomp/sdc/be/config/DistributionEngineConfiguration.java
@@ -32,6 +32,7 @@ public class DistributionEngineConfiguration extends BasicConfiguration {
     private List<String> uebServers;
     private String distributionNotifTopicName;
     private String distributionStatusTopicName;
+    private String kafkaBootStrapServers;
     private Integer initRetryIntervalSec;
     private Integer initMaxIntervalSec;
     private ComponentArtifactTypesConfig distribNotifServiceArtifactTypes;
@@ -75,6 +76,14 @@ public class DistributionEngineConfiguration extends BasicConfiguration {
         this.distributionStatusTopicName = distributionStatusTopicName;
     }
 
+    public String getKafkaBootStrapServers() {
+        return kafkaBootStrapServers;
+    }
+
+    public void setKafkaBootStrapServers(String kafkaBootStrapServers) {
+        this.kafkaBootStrapServers = kafkaBootStrapServers;
+    }
+
     public Integer getInitRetryIntervalSec() {
         return initRetryIntervalSec;
     }
diff --git a/common-app-api/src/test/resources/config/common/distribution-engine-configuration.yaml b/common-app-api/src/test/resources/config/common/distribution-engine-configuration.yaml
index b4aaef59c7..7af7d60867 100644
--- a/common-app-api/src/test/resources/config/common/distribution-engine-configuration.yaml
+++ b/common-app-api/src/test/resources/config/common/distribution-engine-configuration.yaml
@@ -10,6 +10,8 @@ uebSecretKey: ffff
 distributionNotifTopicName: ASDC-DISTR-NOTIF-TOPIC
 distributionStatusTopicName: ASDC-DISTR-STATUS-TOPIC
 
+kafkaBootStrapServers: kafka-bootstrap:9092
+
 initRetryIntervalSec: 5
 initMaxIntervalSec: 60
 
diff --git a/docs/configuration.rst b/docs/configuration.rst
index 54c1760d8f..20d164bfa6 100644
--- a/docs/configuration.rst
+++ b/docs/configuration.rst
@@ -59,6 +59,17 @@ environment.json
                 "fqdn": ["10.0.11.1", "10.0.11.1"]
             },
 
+            # Kafka config
+            "Kafka": {
+             "bootstrap": "kafka-bootstrap:9092"
+            },
+
+            # Messaging topics to be used by clients
+            "DistributionTopics": {
+             "notificationTopicName": "SDC-DISTR-NOTIF-TOPIC",
+             "statusTopicName": "SDC-DISTR-STATUS-TOPIC"
+            },
+
             # IPs used for docker configuration
             "Nodes": {
                 "CS": ["yyy"],
diff --git a/docs/swagger/swagger-sdce-6.json b/docs/swagger/swagger-sdce-6.json
index a950790dbe..394bd5c64b 100644
--- a/docs/swagger/swagger-sdce-6.json
+++ b/docs/swagger/swagger-sdce-6.json
@@ -1,99 +1,111 @@
 {
-  "openapi" : "3.0.1",
-  "info" : {
-    "contact" : {
-      "email" : "onap-discuss@lists.onap.org",
-      "name" : "ONAP",
-      "url" : "https://onap.readthedocs.io"
+  "openapi":"3.0.1",
+  "info":{
+    "contact":{
+      "email":"onap-discuss@lists.onap.org",
+      "name":"ONAP",
+      "url":"https://onap.readthedocs.io"
     },
-    "description" : "SDC API for distribution subscription (SDCE-6)",
-    "license" : {
-      "name" : "Apache 2.0",
-      "url" : "http://www.apache.org/licenses/LICENSE-2.0"
+    "description":"SDC API for distribution subscription (SDCE-6)",
+    "license":{
+      "name":"Apache 2.0",
+      "url":"http://www.apache.org/licenses/LICENSE-2.0"
     },
-    "title" : "SPC API: SDCE-6",
-    "version" : "1.0"
+    "title":"SPC API: SDCE-6",
+    "version":"1.0"
   },
-  "servers" : [ {
-    "description" : "SDCE-6 APIs",
-    "url" : "/sdc"
-  } ],
-  "paths" : {
-    "/v1/artifactTypes" : {
-      "get" : {
-        "description" : "Artifact types list",
-        "operationId" : "getValidArtifactTypes",
-        "parameters" : [ {
-          "description" : "X-ECOMP-RequestID header",
-          "in" : "header",
-          "name" : "X-ECOMP-RequestID",
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "X-ECOMP-InstanceID header",
-          "in" : "header",
-          "name" : "X-ECOMP-InstanceID",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "The username and password",
-          "in" : "header",
-          "name" : "Authorization",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "The username and password",
-          "in" : "header",
-          "name" : "Accept",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
+  "servers":[
+    {
+      "description":"SDCE-6 APIs",
+      "url":"/sdc"
+    }
+  ],
+  "paths":{
+    "/v1/artifactTypes":{
+      "get":{
+        "description":"Artifact types list",
+        "operationId":"getValidArtifactTypes",
+        "parameters":[
+          {
+            "description":"X-ECOMP-RequestID header",
+            "in":"header",
+            "name":"X-ECOMP-RequestID",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"X-ECOMP-InstanceID header",
+            "in":"header",
+            "name":"X-ECOMP-InstanceID",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"The username and password",
+            "in":"header",
+            "name":"Authorization",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"The username and password",
+            "in":"header",
+            "name":"Accept",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
           }
-        } ],
-        "responses" : {
-          "200" : {
-            "content" : {
-              "application/json" : {
-                "schema" : {
-                  "type" : "array",
-                  "items" : {
-                    "type" : "string"
+        ],
+        "responses":{
+          "200":{
+            "content":{
+              "application/json":{
+                "schema":{
+                  "type":"array",
+                  "items":{
+                    "type":"string"
                   }
                 }
               }
             },
-            "description" : "Artifact types list fetched successfully"
+            "description":"Artifact types list fetched successfully"
           },
-          "400" : {
-            "description" : "Missing  'X-ECOMP-InstanceID'  HTTP header - POL5001"
+          "400":{
+            "description":"Missing  'X-ECOMP-InstanceID'  HTTP header - POL5001"
           },
-          "401" : {
-            "description" : "ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
+          "401":{
+            "description":"ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
           },
-          "403" : {
-            "description" : "ECOMP component is not authorized - POL5003"
+          "403":{
+            "description":"ECOMP component is not authorized - POL5003"
           },
-          "405" : {
-            "description" : "Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( POST,PUT,DELETE  will be rejected) - POL4050"
+          "405":{
+            "description":"Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( POST,PUT,DELETE  will be rejected) - POL4050"
           },
-          "500" : {
-            "description" : "The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
+          "500":{
+            "description":"The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
           }
         },
-        "servers" : [ {
-          "url" : "/sdc",
-          "variables" : { }
-        } ],
-        "summary" : "Fetches available artifact types list",
-        "tags" : [ "SDCE-6 APIs" ]
+        "servers":[
+          {
+            "url":"/sdc",
+            "variables":{
+
+            }
+          }
+        ],
+        "summary":"Fetches available artifact types list",
+        "tags":[
+          "SDCE-6 APIs"
+        ]
       }
-    },
-    "/v1/distributionUebCluster" : {
+    },"/v1/distributionUebCluster" : {
       "get" : {
         "description" : "UEB Server List",
         "operationId" : "getUebServerList",
@@ -166,251 +178,393 @@
         "tags" : [ "SDCE-6 APIs" ]
       }
     },
-    "/v1/registerForDistribution" : {
-      "post" : {
-        "description" : "Subscription status",
-        "operationId" : "registerForDistribution",
-        "parameters" : [ {
-          "description" : "X-ECOMP-RequestID header",
-          "in" : "header",
-          "name" : "X-ECOMP-RequestID",
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "X-ECOMP-InstanceID header",
-          "in" : "header",
-          "name" : "X-ECOMP-InstanceID",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "Determines the format of the body of the response",
-          "in" : "header",
-          "name" : "Accept",
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "Determines the format of the body of the request",
-          "in" : "header",
-          "name" : "Content-Type",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "Length  of  the request body",
-          "in" : "header",
-          "name" : "Content-Length",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "The username and password",
-          "in" : "header",
-          "name" : "Authorization",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
+    "/v1/distributionKafkaData":{
+      "get":{
+        "description":"Kafka bootstrap server and topic list",
+        "operationId":"getKafkaData",
+        "parameters":[
+          {
+            "description":"X-ECOMP-RequestID header",
+            "in":"header",
+            "name":"X-ECOMP-RequestID",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"X-ECOMP-InstanceID header",
+            "in":"header",
+            "name":"X-ECOMP-InstanceID",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Determines the format of the body of the response",
+            "in":"header",
+            "name":"Accept",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"The username and password",
+            "in":"header",
+            "name":"Authorization",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
           }
-        } ],
-        "responses" : {
-          "200" : {
-            "content" : {
-              "application/json" : {
-                "schema" : {
-                  "type" : "array",
-                  "items" : {
-                    "$ref" : "#/components/schemas/TopicRegistrationResponse"
+        ],
+        "responses":{
+          "200":{
+            "content":{
+              "application/json":{
+                "schema":{
+                  "type":"array",
+                  "items":{
+                    "$ref":"#/components/schemas/KafkaDataResponse"
                   }
                 }
               }
             },
-            "description" : "ECOMP component is successfully registered for distribution"
+            "description":"ECOMP component is authenticated and kafka endpoint and topic list is returned"
           },
-          "400" : {
-            "description" : "Invalid Body :  Specified 'distrEnvName' doesn’t exist - POL4137"
+          "400":{
+            "description":"Missing  'X-ECOMP-InstanceID'  HTTP header - POL5001"
           },
-          "401" : {
-            "description" : "ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
+          "401":{
+            "description":"ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its credentials  for  Basic Authentication - POL5002"
           },
-          "403" : {
-            "description" : "ECOMP component is not authorized - POL5003"
+          "403":{
+            "description":"ECOMP component is not authorized - POL5003"
           },
-          "405" : {
-            "description" : "Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( PUT,DELETE,GET  will be rejected) - POL4050"
+          "405":{
+            "description":"Method  Not Allowed: Invalid HTTP method type used ( PUT,DELETE,POST will be rejected) - POL4050"
           },
-          "500" : {
-            "description" : "The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
+          "500":{
+            "description":"The GET request failed either due to internal SDC problem or Cambria Service failure. ECOMP Component should continue the attempts to get the needed information - POL5000"
           }
         },
-        "servers" : [ {
-          "url" : "/sdc",
-          "variables" : { }
-        } ],
-        "summary" : "Subscribes for distribution notifications",
-        "tags" : [ "SDCE-6 APIs" ]
+        "servers":[
+          {
+            "url":"/sdc",
+            "variables":{
+
+            }
+          }
+        ],
+        "summary":"return the Kafka bootstrap server and topic list",
+        "tags":[
+          "SDCE-6 APIs"
+        ]
       }
     },
-    "/v1/unRegisterForDistribution" : {
-      "post" : {
-        "description" : "Subscription status",
-        "operationId" : "unRegisterForDistribution",
-        "parameters" : [ {
-          "description" : "X-ECOMP-RequestID header",
-          "in" : "header",
-          "name" : "X-ECOMP-RequestID",
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "X-ECOMP-InstanceID header",
-          "in" : "header",
-          "name" : "X-ECOMP-InstanceID",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
-          }
-        }, {
-          "description" : "Determines the format of the body of the response",
-          "in" : "header",
-          "name" : "Accept",
-          "schema" : {
-            "type" : "string"
+    "/v1/registerForDistribution":{
+      "post":{
+        "description":"Subscription status",
+        "operationId":"registerForDistribution",
+        "parameters":[
+          {
+            "description":"X-ECOMP-RequestID header",
+            "in":"header",
+            "name":"X-ECOMP-RequestID",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"X-ECOMP-InstanceID header",
+            "in":"header",
+            "name":"X-ECOMP-InstanceID",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Determines the format of the body of the response",
+            "in":"header",
+            "name":"Accept",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Determines the format of the body of the request",
+            "in":"header",
+            "name":"Content-Type",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Length  of  the request body",
+            "in":"header",
+            "name":"Content-Length",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"The username and password",
+            "in":"header",
+            "name":"Authorization",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
           }
-        }, {
-          "description" : "Determines the format of the body of the request",
-          "in" : "header",
-          "name" : "Content-Type",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
+        ],
+        "responses":{
+          "200":{
+            "content":{
+              "application/json":{
+                "schema":{
+                  "type":"array",
+                  "items":{
+                    "$ref":"#/components/schemas/TopicRegistrationResponse"
+                  }
+                }
+              }
+            },
+            "description":"ECOMP component is successfully registered for distribution"
+          },
+          "400":{
+            "description":"Invalid Body :  Specified 'distrEnvName' doesn’t exist - POL4137"
+          },
+          "401":{
+            "description":"ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
+          },
+          "403":{
+            "description":"ECOMP component is not authorized - POL5003"
+          },
+          "405":{
+            "description":"Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( PUT,DELETE,GET  will be rejected) - POL4050"
+          },
+          "500":{
+            "description":"The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
           }
-        }, {
-          "description" : "Length  of  the request body",
-          "in" : "header",
-          "name" : "Content-Length",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
+        },
+        "servers":[
+          {
+            "url":"/sdc",
+            "variables":{
+
+            }
           }
-        }, {
-          "description" : "The username and password",
-          "in" : "header",
-          "name" : "Authorization",
-          "required" : true,
-          "schema" : {
-            "type" : "string"
+        ],
+        "summary":"Subscribes for distribution notifications",
+        "tags":[
+          "SDCE-6 APIs"
+        ]
+      }
+    },
+    "/v1/unRegisterForDistribution":{
+      "post":{
+        "description":"Subscription status",
+        "operationId":"unRegisterForDistribution",
+        "parameters":[
+          {
+            "description":"X-ECOMP-RequestID header",
+            "in":"header",
+            "name":"X-ECOMP-RequestID",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"X-ECOMP-InstanceID header",
+            "in":"header",
+            "name":"X-ECOMP-InstanceID",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Determines the format of the body of the response",
+            "in":"header",
+            "name":"Accept",
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Determines the format of the body of the request",
+            "in":"header",
+            "name":"Content-Type",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"Length  of  the request body",
+            "in":"header",
+            "name":"Content-Length",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
+          },
+          {
+            "description":"The username and password",
+            "in":"header",
+            "name":"Authorization",
+            "required":true,
+            "schema":{
+              "type":"string"
+            }
           }
-        } ],
-        "responses" : {
-          "204" : {
-            "content" : {
-              "application/json" : {
-                "schema" : {
-                  "type" : "array",
-                  "items" : {
-                    "$ref" : "#/components/schemas/TopicUnregistrationResponse"
+        ],
+        "responses":{
+          "204":{
+            "content":{
+              "application/json":{
+                "schema":{
+                  "type":"array",
+                  "items":{
+                    "$ref":"#/components/schemas/TopicUnregistrationResponse"
                   }
                 }
               }
             },
-            "description" : "ECOMP component is successfully unregistered"
+            "description":"ECOMP component is successfully unregistered"
           },
-          "400" : {
-            "description" : "Invalid Body :  Specified 'distrEnvName' doesn’t exist - POL4137"
+          "400":{
+            "description":"Invalid Body :  Specified 'distrEnvName' doesn’t exist - POL4137"
           },
-          "401" : {
-            "description" : "ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
+          "401":{
+            "description":"ECOMP component  should authenticate itself  and  to  re-send  again  HTTP  request  with its Basic Authentication credentials - POL5002"
           },
-          "403" : {
-            "description" : "ECOMP component is not authorized - POL5003"
+          "403":{
+            "description":"ECOMP component is not authorized - POL5003"
           },
-          "405" : {
-            "description" : "Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( PUT,DELETE,GET will be rejected) - POL4050"
+          "405":{
+            "description":"Method  Not Allowed  :  Invalid HTTP method type used to  register for  distribution ( PUT,DELETE,GET will be rejected) - POL4050"
           },
-          "500" : {
-            "description" : "The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
+          "500":{
+            "description":"The registration failed due to internal SDC problem or Cambria Service failure ECOMP Component  should  continue the attempts to  register for  distribution - POL5000"
           }
         },
-        "servers" : [ {
-          "url" : "/sdc",
-          "variables" : { }
-        } ],
-        "summary" : "Removes from subscription for distribution notifications",
-        "tags" : [ "SDCE-6 APIs" ]
+        "servers":[
+          {
+            "url":"/sdc",
+            "variables":{
+
+            }
+          }
+        ],
+        "summary":"Removes from subscription for distribution notifications",
+        "tags":[
+          "SDCE-6 APIs"
+        ]
       }
     }
   },
-  "components" : {
-    "schemas" : {
-      "RegistrationRequest" : {
-        "type" : "object",
-        "properties" : {
-          "apiPublicKey" : {
-            "type" : "string"
+  "components":{
+    "schemas":{
+      "RegistrationRequest":{
+        "type":"object",
+        "properties":{
+          "apiPublicKey":{
+            "type":"string"
           },
-          "distEnvEndPoints" : {
-            "type" : "array",
-            "items" : {
-              "type" : "string"
+          "distEnvEndPoints":{
+            "type":"array",
+            "items":{
+              "type":"string"
             }
           },
-          "distrEnvName" : {
-            "type" : "string"
+          "distrEnvName":{
+            "type":"string"
           },
-          "isConsumerToSdcDistrStatusTopic" : {
-            "type" : "boolean"
+          "isConsumerToSdcDistrStatusTopic":{
+            "type":"boolean"
           },
-          "managerApiPublicKey" : {
-            "type" : "string"
+          "managerApiPublicKey":{
+            "type":"string"
           },
-          "managerApiSecretKey" : {
-            "type" : "string"
+          "managerApiSecretKey":{
+            "type":"string"
           }
         }
       },
-      "ServerListResponse" : {
-        "type" : "object",
-        "properties" : {
-          "uebServerList" : {
-            "type" : "array",
-            "items" : {
-              "type" : "string"
+      "ServerListResponse":{
+        "type":"object",
+        "properties":{
+          "uebServerList":{
+            "type":"array",
+            "items":{
+              "type":"string"
             }
           }
         }
       },
-      "TopicRegistrationResponse" : {
-        "type" : "object",
-        "properties" : {
-          "distrNotificationTopicName" : {
-            "type" : "string"
+      "KafkaDataResponse":{
+        "type":"object",
+        "properties":{
+          "kafkaBootStrapServer":{
+            "type":"string"
           },
-          "distrStatusTopicName" : {
-            "type" : "string"
+          "distrNotificationTopicName":{
+            "type":"string"
+          },
+          "distrStatusTopicName":{
+            "type":"string"
           }
         }
       },
-      "TopicUnregistrationResponse" : {
-        "type" : "object",
-        "properties" : {
-          "distrNotificationTopicName" : {
-            "type" : "string"
+      "TopicRegistrationResponse":{
+        "type":"object",
+        "properties":{
+          "distrNotificationTopicName":{
+            "type":"string"
           },
-          "distrStatusTopicName" : {
-            "type" : "string"
+          "distrStatusTopicName":{
+            "type":"string"
+          }
+        }
+      },
+      "TopicUnregistrationResponse":{
+        "type":"object",
+        "properties":{
+          "distrNotificationTopicName":{
+            "type":"string"
+          },
+          "distrStatusTopicName":{
+            "type":"string"
           },
-          "notificationUnregisterResult" : {
-            "type" : "string",
-            "enum" : [ "OK", "CONNNECTION_ERROR", "NOT_FOUND", "TOPIC_ALREADY_EXIST", "OBJECT_NOT_FOUND", "INTERNAL_SERVER_ERROR", "AUTHENTICATION_ERROR", "UNKNOWN_HOST_ERROR" ]
+          "notificationUnregisterResult":{
+            "type":"string",
+            "enum":[
+              "OK",
+              "CONNNECTION_ERROR",
+              "NOT_FOUND",
+              "TOPIC_ALREADY_EXIST",
+              "OBJECT_NOT_FOUND",
+              "INTERNAL_SERVER_ERROR",
+              "AUTHENTICATION_ERROR",
+              "UNKNOWN_HOST_ERROR"
+            ]
           },
-          "statusUnregisterResult" : {
-            "type" : "string",
-            "enum" : [ "OK", "CONNNECTION_ERROR", "NOT_FOUND", "TOPIC_ALREADY_EXIST", "OBJECT_NOT_FOUND", "INTERNAL_SERVER_ERROR", "AUTHENTICATION_ERROR", "UNKNOWN_HOST_ERROR" ]
+          "statusUnregisterResult":{
+            "type":"string",
+            "enum":[
+              "OK",
+              "CONNNECTION_ERROR",
+              "NOT_FOUND",
+              "TOPIC_ALREADY_EXIST",
+              "OBJECT_NOT_FOUND",
+              "INTERNAL_SERVER_ERROR",
+              "AUTHENTICATION_ERROR",
+              "UNKNOWN_HOST_ERROR"
+            ]
           }
         }
       }
diff --git a/integration-tests/environments/integration-test.json b/integration-tests/environments/integration-test.json
index 3402113ed2..c2c50d1548 100644
--- a/integration-tests/environments/integration-test.json
+++ b/integration-tests/environments/integration-test.json
@@ -27,6 +27,12 @@
          "SecretKey": "Ehq3WyT4bkif4zwgEbvshGal",
          "fqdn": ["10.0.11.1", "10.0.11.1"]
       },
+      "Kafka": {
+         "bootstrap": "kafka-bootstrap-int:9092"
+      },
+      "DistributionTopics": {
+         "notificationTopicName": "SDC-DISTR-NOTIF-TOPIC-INT"
+      },
       "Nodes": {
          "CS": [
             "sdc-cs"
diff --git a/sdc-os-chef/environments/Template.json b/sdc-os-chef/environments/Template.json
index b0d908c93d..7cd95cc3f9 100644
--- a/sdc-os-chef/environments/Template.json
+++ b/sdc-os-chef/environments/Template.json
@@ -27,6 +27,13 @@
          "SecretKey": "Ehq3WyT4bkif4zwgEbvshGal",
          "fqdn": ["10.0.11.1", "10.0.11.1"]
       },
+      "Kafka": {
+         "bootstrap": "kafka-bootstrap:9092"
+      },
+      "DistributionTopics": {
+         "notificationTopicName": "SDC-DISTR-NOTIF-TOPIC",
+         "statusTopicName": "SDC-DISTR-STATUS-TOPIC"
+      },
       "Nodes": {
          "CS": [
             "yyy"