Disable DMaaP if Kafka active

DMaaP healthcheck cannot succeed when using Kafka, which in turn causes distribution to fail

Signed-off-by: Vasyl Razinkov <vasyl.razinkov@est.tech>
Change-Id: Ia6e1e35b9a2e819e38e62caeb797342948f34e92
Issue-ID: SDC-4542

12 files changed, 380 insertions, 355 deletions

diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngine.java b/catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngine.java
index cb14ebeebe..cebedadf10 100644
--- a/catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngine.java
+++ b/catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngine.java
@@ -19,18 +19,10 @@
  */
 package org.openecomp.sdc.be.components.distribution.engine;
 
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import javax.annotation.Resource;
+import lombok.Setter;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.openecomp.sdc.be.components.kafka.KafkaHandler;
 import org.openecomp.sdc.be.components.validation.ServiceDistributionValidation;
 import org.openecomp.sdc.be.config.BeEcompErrorManager;
 import org.openecomp.sdc.be.config.ConfigurationManager;
@@ -44,15 +36,29 @@ import org.openecomp.sdc.common.log.wrappers.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
 @Component("distributionEngine")
 public class DistributionEngine implements IDistributionEngine {
 
     private static final Logger logger = Logger.getLogger(DistributionEngine.class.getName());
     private static final Pattern FQDN_PATTERN = Pattern.compile(
-        "^([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])(\\.([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9]))*(:[0-9]{2,4})*$",
-        Pattern.CASE_INSENSITIVE);
+            "^([A-Z0-9]|[A-Z0-9][A-Z0-9\\-]{0,61}[A-Z0-9])(\\.([A-Z0-9]|[A-Z0-9][A-Z0-9\\-]{0,61}[A-Z0-9]))*(:[0-9]{2,4})*$",
+            Pattern.CASE_INSENSITIVE);
     @Autowired
     private EnvironmentsEngine environmentsEngine;
+    @Autowired
+    @Setter
+    private KafkaHandler kafkaHandler;
     @Resource
     private DistributionNotificationSender distributionNotificationSender;
     @Resource
@@ -83,7 +89,7 @@ public class DistributionEngine implements IDistributionEngine {
     private void init() {
         logger.trace("Enter init method of DistributionEngine");
         DistributionEngineConfiguration distributionEngineConfiguration = ConfigurationManager.getConfigurationManager()
-            .getDistributionEngineConfiguration();
+                .getDistributionEngineConfiguration();
         boolean startDistributionEngine = distributionEngineConfiguration.isStartDistributionEngine();
         logger.debug("Distribution engine activation parameter is {}", startDistributionEngine);
         if (!startDistributionEngine) {
@@ -94,7 +100,7 @@ public class DistributionEngine implements IDistributionEngine {
         boolean isValidConfig = validateConfiguration(distributionEngineConfiguration);
         if (!isValidConfig) {
             BeEcompErrorManager.getInstance()
-                .logBeUebSystemError(DistributionEngineInitTask.INIT_DISTRIBUTION_ENGINE_FLOW, "validate distribution configuration in init phase");
+                    .logBeUebSystemError(DistributionEngineInitTask.INIT_DISTRIBUTION_ENGINE_FLOW, "validate distribution configuration in init phase");
             this.distributionEngineClusterHealth.setHealthCheckUebConfigurationError();
             return;
         }
@@ -133,13 +139,16 @@ public class DistributionEngine implements IDistributionEngine {
      */
     protected boolean validateConfiguration(DistributionEngineConfiguration deConfiguration) {
         String methodName = "validateConfiguration";
-        boolean result = isValidServers(deConfiguration.getUebServers(), methodName, "uebServers");
-        result = isValidParam(deConfiguration.getEnvironments(), methodName, "environments") && result;
-        result = isValidParam(deConfiguration.getUebPublicKey(), methodName, "uebPublicKey") && result;
-        result = isValidParam(deConfiguration.getUebSecretKey(), methodName, "uebSecretKey") && result;
+        boolean result = isValidParam(deConfiguration.getEnvironments(), methodName, "environments");
+
+        if (!kafkaHandler.isKafkaActive()) {
+            result = isValidServers(deConfiguration.getUebServers(), methodName, "uebServers") && result;
+            result = isValidParam(deConfiguration.getUebPublicKey(), methodName, "uebPublicKey") && result;
+            result = isValidParam(deConfiguration.getUebSecretKey(), methodName, "uebSecretKey") && result;
+            result = isValidObject(deConfiguration.getCreateTopic(), methodName, "createTopic") && result;
+        }
         result = isValidParam(deConfiguration.getDistributionNotifTopicName(), methodName, "distributionNotifTopicName") && result;
         result = isValidParam(deConfiguration.getDistributionStatusTopicName(), methodName, "distributionStatusTopicName") && result;
-        result = isValidObject(deConfiguration.getCreateTopic(), methodName, "createTopic") && result;
         result = isValidObject(deConfiguration.getDistributionStatusTopic(), methodName, "distributionStatusTopic") && result;
         result = isValidObject(deConfiguration.getInitMaxIntervalSec(), methodName, "initMaxIntervalSec") && result;
         result = isValidObject(deConfiguration.getInitRetryIntervalSec(), methodName, "initRetryIntervalSec") && result;
@@ -251,13 +260,13 @@ public class DistributionEngine implements IDistributionEngine {
     public ActionStatus notifyService(String distributionId, Service service, INotificationData notificationData, String envId, String envName,
                                       User modifier) {
         logger.debug(
-            "Received notify service request. distributionId = {}, serviceUuid = {} serviceUid = {}, envName = {}, userId = {}, modifierName {}",
-            distributionId, service.getUUID(), service.getUniqueId(), envName, service.getLastUpdaterUserId(), modifier);
+                "Received notify service request. distributionId = {}, serviceUuid = {} serviceUid = {}, envName = {}, userId = {}, modifierName {}",
+                distributionId, service.getUUID(), service.getUniqueId(), envName, service.getLastUpdaterUserId(), modifier);
         String topicName = buildTopicName(envName);
         ActionStatus notifyServiceStatus = Optional.ofNullable(environmentsEngine.getEnvironmentById(envId)).map(EnvironmentMessageBusData::new).map(
-            messageBusData -> distributionNotificationSender
-                .sendNotification(topicName, distributionId, messageBusData, notificationData, service, modifier))
-            .orElse(ActionStatus.DISTRIBUTION_ENVIRONMENT_NOT_AVAILABLE);
+                        messageBusData -> distributionNotificationSender
+                                .sendNotification(topicName, distributionId, messageBusData, notificationData, service, modifier))
+                .orElse(ActionStatus.DISTRIBUTION_ENVIRONMENT_NOT_AVAILABLE);
         logger.debug("Finish notifyService. status is {}", notifyServiceStatus);
         return notifyServiceStatus;
     }
@@ -274,7 +283,7 @@ public class DistributionEngine implements IDistributionEngine {
         if (status != StorageOperationStatus.OK) {
             String envErrorDec = getEnvironmentErrorDescription(status);
             BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError(DistributionNotificationSender.DISTRIBUTION_NOTIFICATION_SENDING,
-                "Environment name " + envName + " is not available. Reason : " + envErrorDec);
+                    "Environment name " + envName + " is not available. Reason : " + envErrorDec);
         }
         return status;
     }
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogic.java b/catalog-be/src/main/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogic.java
index cacbd1369d..e641365afa 100644
--- a/catalog-be/src/main/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogic.java
+++ b/catalog-be/src/main/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogic.java
@@ -69,6 +69,7 @@ import org.openecomp.sdc.be.components.health.HealthCheckBusinessLogic;
 import org.openecomp.sdc.be.components.impl.exceptions.ByActionStatusComponentException;
 import org.openecomp.sdc.be.components.impl.exceptions.ByResponseFormatComponentException;
 import org.openecomp.sdc.be.components.impl.exceptions.ComponentException;
+import org.openecomp.sdc.be.components.kafka.KafkaHandler;
 import org.openecomp.sdc.be.components.path.ForwardingPathValidator;
 import org.openecomp.sdc.be.components.utils.InterfaceOperationUtils;
 import org.openecomp.sdc.be.components.utils.PropertiesUtils;
@@ -176,7 +177,7 @@ import org.springframework.web.context.WebApplicationContext;
 @org.springframework.stereotype.Component("serviceBusinessLogic")
 public class ServiceBusinessLogic extends ComponentBusinessLogic {
 
-    static final String IS_VALID = "isValid";
+    private static final String IS_VALID = "isValid";
     private static final String THE_SERVICE_WITH_SYSTEM_NAME_LOCKED = "The service with system name {} locked. ";
     private static final String FAILED_TO_LOCK_SERVICE_RESPONSE_IS = "Failed to lock service {}. Response is {}. ";
     private static final String AUDIT_BEFORE_SENDING_RESPONSE = "audit before sending response";
@@ -196,6 +197,7 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
     private final ServiceCategoryValidator serviceCategoryValidator;
     private final ServiceValidator serviceValidator;
     private final GroupBusinessLogic groupBusinessLogic;
+    private final KafkaHandler kafkaHandler;
     private ForwardingPathOperation forwardingPathOperation;
     private AuditCassandraDao auditCassandraDao;
     private ServiceTypeValidator serviceTypeValidator;
@@ -214,7 +216,7 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
                                 ComponentDescriptionValidator componentDescriptionValidator, ModelOperation modelOperation,
                                 final ServiceRoleValidator serviceRoleValidator,
                                 final ServiceInstantiationTypeValidator serviceInstantiationTypeValidator,
-                                final ServiceCategoryValidator serviceCategoryValidator, final ServiceValidator serviceValidator) {
+                                final ServiceCategoryValidator serviceCategoryValidator, final ServiceValidator serviceValidator, KafkaHandler kafkaHandler) {
         super(elementDao, groupOperation, groupInstanceOperation, groupTypeOperation, groupBusinessLogic, interfaceOperation,
             interfaceLifecycleTypeOperation, artifactsBusinessLogic, artifactToscaOperation, componentContactIdValidator, componentNameValidator,
             componentTagsValidator, componentValidator, componentIconValidator, componentProjectCodeValidator, componentDescriptionValidator);
@@ -229,6 +231,7 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
         this.serviceCategoryValidator = serviceCategoryValidator;
         this.serviceValidator = serviceValidator;
         this.groupBusinessLogic = groupBusinessLogic;
+        this.kafkaHandler = kafkaHandler;
     }
 
     @Autowired
@@ -1730,22 +1733,24 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
             log.trace("Update environment name to be {} instead of {}", configuredEnvName, envName);
             envName = configuredEnvName;
         }
-        // DE194021
-        ServletContext servletContext = request.getSession().getServletContext();
-        boolean isDistributionEngineUp = getHealthCheckBL(servletContext).isDistributionEngineUp(); // DE
-        if (!isDistributionEngineUp) {
-            BeEcompErrorManager.getInstance().logBeSystemError("Distribution Engine is DOWN");
-            log.debug("Distribution Engine is DOWN");
-            response = componentsUtils.getResponseFormat(ActionStatus.GENERAL_ERROR);
-            return Either.right(response);
+        if (!kafkaHandler.isKafkaActive()) {
+            // DE194021
+            ServletContext servletContext = request.getSession().getServletContext();
+            boolean isDistributionEngineUp = getHealthCheckBL(servletContext).isDistributionEngineUp(); // DE
+            if (!isDistributionEngineUp) {
+                BeEcompErrorManager.getInstance().logBeSystemError("Distribution Engine is DOWN");
+                log.debug("Distribution Engine is DOWN");
+                response = componentsUtils.getResponseFormat(ActionStatus.GENERAL_ERROR);
+                return Either.right(response);
+            }
         }
         Either<Service, StorageOperationStatus> serviceRes = toscaOperationFacade.getToscaElement(serviceId);
         if (serviceRes.isRight()) {
             log.debug("failed retrieving service");
             response = componentsUtils
-                .getResponseFormat(componentsUtils.convertFromStorageResponse(serviceRes.right().value(), ComponentTypeEnum.SERVICE), serviceId);
+                    .getResponseFormat(componentsUtils.convertFromStorageResponse(serviceRes.right().value(), ComponentTypeEnum.SERVICE), serviceId);
             componentsUtils.auditComponent(response, user, null, AuditingActionEnum.DISTRIBUTION_STATE_CHANGE_REQUEST,
-                new ResourceCommonInfo(ComponentTypeEnum.SERVICE.getValue()), ResourceVersionInfo.newBuilder().build(), did);
+                    new ResourceCommonInfo(ComponentTypeEnum.SERVICE.getValue()), ResourceVersionInfo.newBuilder().build(), did);
             return Either.right(response);
         }
         Service service = serviceRes.left().value();
@@ -1756,7 +1761,7 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
         if (service.getLifecycleState() != LifecycleStateEnum.CERTIFIED) {
             log.info("service {} is  not available for distribution. Should be in certified state", service.getUniqueId());
             ResponseFormat responseFormat = componentsUtils
-                .getResponseFormat(ActionStatus.SERVICE_NOT_AVAILABLE_FOR_DISTRIBUTION, service.getVersion(), service.getName());
+                    .getResponseFormat(ActionStatus.SERVICE_NOT_AVAILABLE_FOR_DISTRIBUTION, service.getVersion(), service.getName());
             return Either.right(responseFormat);
         }
         String dcurrStatus = service.getDistributionStatus().name();
@@ -1767,7 +1772,7 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
             ActionStatus notifyServiceResponse = distributionEngine.notifyService(did, service, notificationData, envName, user);
             if (notifyServiceResponse == ActionStatus.OK) {
                 Either<Service, ResponseFormat> updateStateRes = updateDistributionStatusForActivation(service, user,
-                    DistributionStatusEnum.DISTRIBUTED);
+                        DistributionStatusEnum.DISTRIBUTED);
                 if (updateStateRes.isLeft() && updateStateRes.left().value() != null) {
                     updatedService = updateStateRes.left().value();
                     updatedStatus = updatedService.getDistributionStatus().name();
@@ -1786,13 +1791,13 @@ public class ServiceBusinessLogic extends ComponentBusinessLogic {
             }
         } else {
             response = componentsUtils
-                .getResponseFormatByDE(componentsUtils.convertFromStorageResponse(readyForDistribution, ComponentTypeEnum.SERVICE), envName);
+                    .getResponseFormatByDE(componentsUtils.convertFromStorageResponse(readyForDistribution, ComponentTypeEnum.SERVICE), envName);
             result = Either.right(response);
         }
         componentsUtils.auditComponent(response, user, service, AuditingActionEnum.DISTRIBUTION_STATE_CHANGE_REQUEST,
-            new ResourceCommonInfo(service.getName(), ComponentTypeEnum.SERVICE.getValue()),
-            ResourceVersionInfo.newBuilder().distributionStatus(dcurrStatus).build(),
-            ResourceVersionInfo.newBuilder().distributionStatus(updatedStatus).build(), null, null, did);
+                new ResourceCommonInfo(service.getName(), ComponentTypeEnum.SERVICE.getValue()),
+                ResourceVersionInfo.newBuilder().distributionStatus(dcurrStatus).build(),
+                ResourceVersionInfo.newBuilder().distributionStatus(updatedStatus).build(), null, null, did);
         return result;
     }
 
diff --git a/catalog-be/src/main/java/org/openecomp/sdc/config/CatalogBESpringConfig.java b/catalog-be/src/main/java/org/openecomp/sdc/config/CatalogBESpringConfig.java
index b38210116c..5c018fc0f0 100644
--- a/catalog-be/src/main/java/org/openecomp/sdc/config/CatalogBESpringConfig.java
+++ b/catalog-be/src/main/java/org/openecomp/sdc/config/CatalogBESpringConfig.java
@@ -58,6 +58,7 @@ import org.springframework.core.annotation.Order;
     "org.openecomp.sdc.be.components.csar",
     "org.openecomp.sdc.be.components.property",
     "org.openecomp.sdc.be.components.attribute",
+    "org.openecomp.sdc.be.components.kafka",
     "org.openecomp.sdc.be.csar.security",
     "org.openecomp.sdc.be.datamodel.utils",
     "org.openecomp.sdc.be.components.upgrade",
diff --git a/catalog-be/src/main/resources/application-context.xml b/catalog-be/src/main/resources/application-context.xml
index 39ac4f8892..2b1d5ef3bd 100644
--- a/catalog-be/src/main/resources/application-context.xml
+++ b/catalog-be/src/main/resources/application-context.xml
@@ -15,6 +15,7 @@
                          org.openecomp.sdc.be.externalapi.servlet,
                          org.openecomp.sdc.be.components.scheduledtasks,
                          org.openecomp.sdc.be.facade.operations,
+                         org.openecomp.sdc.be.components.kafka,
                          org.openecomp.sdc.be.components.impl">
   </context:component-scan>
 
@@ -23,7 +24,7 @@
   <bean class="org.openecomp.sdc.be.components.distribution.engine.config.DistributionEngineSpringConfig"/>
   <bean class="org.openecomp.sdc.config.CatalogBESpringConfig"/>
   <bean class="org.openecomp.sdc.config.MicrometerSpringConfig"/>
-
+  <bean class="org.openecomp.sdc.be.components.kafka.KafkaHandler"/>
 
   <aop:config>
     <aop:aspect id="lockAspect" ref="componentLockAspect">
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/BaseServiceBusinessLogicTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/BaseServiceBusinessLogicTest.java
index c3fc69895b..638eef6898 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/BaseServiceBusinessLogicTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/BaseServiceBusinessLogicTest.java
@@ -187,7 +187,7 @@ public abstract class BaseServiceBusinessLogicTest extends ComponentBusinessLogi
             serviceDistributionValidation, forwardingPathValidator, uiComponentDataConverter,
             artifactToscaOperation, componentContactIdValidator, componentNameValidator,
             componentTagsValidator, componentValidator, componentIconValidator, componentProjectCodeValidator, componentDescriptionValidator,
-            modelOperation, null, null, null, null);
+            modelOperation, null, null, null, null, null);
         bl.setUserAdmin(mockUserAdmin);
         bl.setGraphLockOperation(graphLockOperation);
         bl.setJanusGraphDao(mockJanusGraphDao);
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/ServiceDistributionBLTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/ServiceDistributionBLTest.java
index e2d4e0ee78..602a3bb4ce 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/ServiceDistributionBLTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/ServiceDistributionBLTest.java
@@ -77,7 +77,7 @@ class ServiceDistributionBLTest extends ComponentBusinessLogicMock {
         artifactToscaOperation, componentContactIdValidator,
         componentNameValidator, componentTagsValidator, componentValidator, componentIconValidator,
         componentProjectCodeValidator, componentDescriptionValidator, modelOperation, null, null,
-        null, null);
+        null, null, null);
 
     private Service serviceToActivate;
     private ActivationRequestInformation activationRequestInformation;
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineConfigTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineConfigTest.java
index a8aa7478fa..21a9b198f1 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineConfigTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineConfigTest.java
@@ -7,9 +7,9 @@
  * 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.
@@ -20,8 +20,12 @@
 
 package org.openecomp.sdc.be.components.distribution.engine;
 
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.openecomp.sdc.be.components.kafka.KafkaHandler;
 import org.openecomp.sdc.be.config.ConfigurationManager;
 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
 import org.openecomp.sdc.be.config.DistributionEngineConfiguration.ComponentArtifactTypesConfig;
@@ -35,24 +39,30 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
 
-public class DistributionEngineConfigTest {
+class DistributionEngineConfigTest {
 
-    @Before
+    @Mock
+    private KafkaHandler kafkaHandler;
+
+    @BeforeEach
     public void setup() {
+        MockitoAnnotations.openMocks(this);
+
         ExternalConfiguration.setAppName("catalog-be");
         ExternalConfiguration.setConfigDir("src/test/resources/config");
         ExternalConfiguration.listenForChanges();
 
-        ConfigurationSource configurationSource = new FSConfigurationSource(ExternalConfiguration.getChangeListener(), ExternalConfiguration.getConfigDir() + File.separator + ExternalConfiguration.getAppName());
-
-        ConfigurationManager configurationManager = new ConfigurationManager(configurationSource);
+        new ConfigurationManager(new FSConfigurationSource(ExternalConfiguration.getChangeListener(), ExternalConfiguration.getConfigDir() + File.separator + ExternalConfiguration.getAppName()));
     }
 
     @Test
-    public void validateMissingEnvironments() {
+    void validateMissingEnvironments() {
+
+        when(kafkaHandler.isKafkaActive()).thenReturn(false);
 
         DistributionEngineConfiguration deConfiguration = new DistributionEngineConfiguration();
 
@@ -60,7 +70,7 @@ public class DistributionEngineConfigTest {
         String uebSecretKey = "uebSecretKey";
 
         DistributionEngine distributionEngine = new DistributionEngine();
-
+        distributionEngine.setKafkaHandler(kafkaHandler);
         List<String> environments = new ArrayList<>();
         environments.add("PROD");
         deConfiguration.setEnvironments(environments);
@@ -97,71 +107,71 @@ public class DistributionEngineConfigTest {
         deConfiguration.setInitRetryIntervalSec(3);
 
         boolean isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setUebServers(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setUebServers(servers);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setEnvironments(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setEnvironments(environments);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setUebPublicKey(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setUebPublicKey(uebPublicKey);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setUebSecretKey(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setUebSecretKey(uebPublicKey);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setDistributionNotifTopicName(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setDistributionNotifTopicName(uebPublicKey);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setDistributionStatusTopicName(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setDistributionStatusTopicName(uebPublicKey);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setInitMaxIntervalSec(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setInitMaxIntervalSec(8);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
         deConfiguration.setInitRetryIntervalSec(null);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertFalse("check empty configuration", isValid);
+        assertFalse(isValid, "check empty configuration");
 
         deConfiguration.setInitRetryIntervalSec(8);
         isValid = distributionEngine.validateConfiguration(deConfiguration);
-        assertTrue("check empty configuration", isValid);
+        assertTrue(isValid, "check empty configuration");
 
     }
 
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineTest.java
index 3e70a3229f..c5d8d37d39 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/distribution/engine/DistributionEngineTest.java
@@ -7,9 +7,9 @@
  * 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.
@@ -21,17 +21,16 @@
 package org.openecomp.sdc.be.components.distribution.engine;
 
 import mockit.Deencapsulation;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
+import org.openecomp.sdc.be.components.kafka.KafkaHandler;
 import org.openecomp.sdc.be.components.utils.OperationalEnvironmentBuilder;
 import org.openecomp.sdc.be.config.Configuration;
 import org.openecomp.sdc.be.config.DistributionEngineConfiguration;
-import org.openecomp.sdc.be.config.DistributionEngineConfiguration.DistributionStatusTopicConfig;
 import org.openecomp.sdc.be.dao.api.ActionStatus;
 import org.openecomp.sdc.be.model.Service;
 import org.openecomp.sdc.be.model.User;
@@ -46,68 +45,69 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
-public class DistributionEngineTest{
+public class DistributionEngineTest {
 
-	public static final String DISTRIBUTION_ID = "distId";
-	public static final String ENV_ID = "envId";
-	public static final String USER_ID = "userId";
-	public static final String MODIFIER = "modifier";
+    public static final String DISTRIBUTION_ID = "distId";
+    public static final String ENV_ID = "envId";
+    public static final String USER_ID = "userId";
+    public static final String MODIFIER = "modifier";
 
-	@InjectMocks
-	private DistributionEngine testInstance;
+    @InjectMocks
+    private DistributionEngine testInstance;
 
-	@Mock
-	private EnvironmentsEngine environmentsEngine;
+    @Mock
+    private EnvironmentsEngine environmentsEngine;
+    @Mock
+    private KafkaHandler kafkaHandler;
+    @Mock
+    private DistributionNotificationSender distributionNotificationSender;
+    @Mock
+    private ServiceDistributionArtifactsBuilder serviceDistributionArtifactsBuilder;
 
-	@Mock
-	private DistributionNotificationSender distributionNotificationSender;
-	
-	@Mock
-	private ServiceDistributionArtifactsBuilder serviceDistributionArtifactsBuilder;
-	
-	private DummyDistributionConfigurationManager distributionEngineConfigurationMock;
+    private DummyDistributionConfigurationManager distributionEngineConfigurationMock;
 
-	private Map<String, OperationalEnvironmentEntry> envs;
+    private Map<String, OperationalEnvironmentEntry> envs;
 
     private User modifier = new User();
 
     private Configuration.EnvironmentContext environmentContext = mock(Configuration.EnvironmentContext.class);
 
-	@Before
-	public void setUpMock() throws Exception {
-		MockitoAnnotations.openMocks(this);
-		distributionEngineConfigurationMock = new DummyDistributionConfigurationManager();
-		envs = getEnvs(ENV_ID);
+    @BeforeEach
+    public void setUpMock() throws Exception {
+        MockitoAnnotations.openMocks(this);
+        distributionEngineConfigurationMock = new DummyDistributionConfigurationManager();
+        envs = getEnvs(ENV_ID);
         modifier.setUserId(USER_ID);
         modifier.setFirstName(MODIFIER);
         modifier.setLastName(MODIFIER);
         when(environmentContext.getDefaultValue()).thenReturn("General_Revenue-Bearing");
         when(distributionEngineConfigurationMock.getConfiguration().getEnvironmentContext())
-				.thenReturn(environmentContext);
-	}
+                .thenReturn(environmentContext);
+    }
 
     @Test
-    public void notifyService() throws Exception {
+    void notifyService() throws Exception {
         NotificationDataImpl notificationData = new NotificationDataImpl();
         Service service = new Service();
         when(environmentsEngine.getEnvironmentById(ENV_ID)).thenReturn(envs.get(ENV_ID));
         when(distributionEngineConfigurationMock.getConfigurationMock().getDistributionNotifTopicName()).thenReturn("topic");
         when(distributionNotificationSender.sendNotification(eq("topic-ENVID"), eq(DISTRIBUTION_ID), any(EnvironmentMessageBusData.class),
                 any(NotificationDataImpl.class), any(Service.class), any(User.class)))
-        .thenReturn(ActionStatus.OK);
+                .thenReturn(ActionStatus.OK);
         ActionStatus actionStatus = testInstance.notifyService(DISTRIBUTION_ID, service, notificationData, ENV_ID, modifier);
         assertEquals(ActionStatus.OK, actionStatus);
     }
 
     @Test
-    public void notifyService_couldNotResolveEnvironment() throws Exception {
+    void notifyService_couldNotResolveEnvironment() throws Exception {
         when(environmentsEngine.getEnvironments()).thenReturn(envs);
         ActionStatus actionStatus = testInstance.notifyService(DISTRIBUTION_ID, new Service(), new NotificationDataImpl(), "someNonExisitngEnv", modifier);
         assertEquals(ActionStatus.DISTRIBUTION_ENVIRONMENT_NOT_AVAILABLE, actionStatus);
@@ -115,7 +115,7 @@ public class DistributionEngineTest{
     }
 
     @Test
-    public void notifyService_failedWhileSendingNotification() throws Exception {
+    void notifyService_failedWhileSendingNotification() throws Exception {
         NotificationDataImpl notificationData = new NotificationDataImpl();
         Service service = new Service();
         when(environmentsEngine.getEnvironmentById(ENV_ID)).thenReturn(envs.get(ENV_ID));
@@ -127,157 +127,160 @@ public class DistributionEngineTest{
         assertEquals(ActionStatus.GENERAL_ERROR, actionStatus);
     }
 
-	private Map<String, OperationalEnvironmentEntry> getEnvs(String... environmentIds) {
-		Set<String> uebAddress = new HashSet<>();
-		uebAddress.add("someAddress");
-		return Stream.of(environmentIds)
-				.map(id -> new OperationalEnvironmentBuilder().setEnvId(id).setDmaapUebAddress(uebAddress).build())
-				.collect(Collectors.toMap(OperationalEnvironmentEntry::getEnvironmentId, Function.identity()));
-	}
+    private Map<String, OperationalEnvironmentEntry> getEnvs(String... environmentIds) {
+        Set<String> uebAddress = new HashSet<>();
+        uebAddress.add("someAddress");
+        return Stream.of(environmentIds)
+                .map(id -> new OperationalEnvironmentBuilder().setEnvId(id).setDmaapUebAddress(uebAddress).build())
+                .collect(Collectors.toMap(OperationalEnvironmentEntry::getEnvironmentId, Function.identity()));
+    }
 
-	private DistributionEngine createTestSubject() {
-		return new DistributionEngine();
-	}
+    private DistributionEngine createTestSubject() {
+        return new DistributionEngine();
+    }
 
-	@Test(expected=NullPointerException.class)
-	public void testInit() throws Exception {
-		DistributionEngine testSubject;
+    @Test
+    public void testInit() throws Exception {
+        DistributionEngine testSubject;
 
-		// default test
-		testSubject = createTestSubject();
-		Deencapsulation.invoke(testSubject, "init");
-	}
+        // default test
+        testSubject = createTestSubject();
+        assertThrows(NullPointerException.class, () -> Deencapsulation.invoke(testSubject, "init"));
+    }
 
-	@Test
-	public void testShutdown() throws Exception {
-		DistributionEngine testSubject;
+    @Test
+    void testShutdown() throws Exception {
+        DistributionEngine testSubject;
 
-		// default test
-		testSubject = createTestSubject();
-		testSubject.shutdown();
-	}
+        // default test
+        testSubject = createTestSubject();
+        testSubject.shutdown();
+    }
 
-	@Test
-	public void testValidateConfiguration() throws Exception {
-		DistributionEngine testSubject;
+    @Test
+    void testValidateConfiguration() throws Exception {
+        when(kafkaHandler.isKafkaActive()).thenReturn(false);
+
+        DistributionEngine testSubject;
 		DistributionEngineConfiguration deConfiguration = new DistributionEngineConfiguration();
-		deConfiguration.setDistributionStatusTopic((new DistributionStatusTopicConfig()));
+		deConfiguration.setDistributionStatusTopic((new DistributionEngineConfiguration.DistributionStatusTopicConfig()));
 		boolean result;
 
 		// default test
 		testSubject = createTestSubject();
+        testSubject.setKafkaHandler(kafkaHandler);
 		result = Deencapsulation.invoke(testSubject, "validateConfiguration", deConfiguration);
-	}
-
-	@Test
-	public void testIsValidServers() throws Exception {
-		DistributionEngine testSubject;
-		List<String> uebServers = null;
-		String methodName = "";
-		String paramName = "";
-		boolean result;
-
-		// test 1
-		testSubject = createTestSubject();
-		uebServers = null;
-		result = Deencapsulation.invoke(testSubject, "isValidServers",
-				new Object[] { List.class, methodName, paramName });
-		Assert.assertEquals(false, result);
-	}
-
-	@Test
-	public void testIsValidFqdn() throws Exception {
-		DistributionEngine testSubject;
-		String serverFqdn = "";
-		boolean result;
-
-		// default test
-		testSubject = createTestSubject();
-		result = Deencapsulation.invoke(testSubject, "isValidFqdn", new Object[] { serverFqdn });
-	}
-
-	@Test
-	public void testIsValidParam() throws Exception {
-		DistributionEngine testSubject;
-		String paramValue = "";
-		String methodName = "";
-		String paramName = "";
-		boolean result;
-
-		// default test
-		testSubject = createTestSubject();
-		result = Deencapsulation.invoke(testSubject, "isValidParam",
-				new Object[] { paramValue, methodName, paramName });
-	}
+    }
 
-	@Test
-	public void testIsValidParam_1() throws Exception {
-		DistributionEngine testSubject;
-		List<String> paramValue = null;
-		String methodName = "";
-		String paramName = "";
-		boolean result;
+    @Test
+    void testIsValidServers() throws Exception {
+        DistributionEngine testSubject;
+        List<String> uebServers = null;
+        String methodName = "";
+        String paramName = "";
+        boolean result;
+
+        // test 1
+        testSubject = createTestSubject();
+        uebServers = null;
+        result = Deencapsulation.invoke(testSubject, "isValidServers",
+                new Object[]{List.class, methodName, paramName});
+        assertEquals(false, result);
+    }
 
-		// default test
-		testSubject = createTestSubject();
-		result = Deencapsulation.invoke(testSubject, "isValidParam",
-				new Object[] { List.class, methodName, paramName });
-	}
+    @Test
+    void testIsValidFqdn() throws Exception {
+        DistributionEngine testSubject;
+        String serverFqdn = "";
+        boolean result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = Deencapsulation.invoke(testSubject, "isValidFqdn", new Object[]{serverFqdn});
+    }
 
-	@Test
-	public void testIsValidObject() throws Exception {
-		DistributionEngine testSubject;
-		Object paramValue = null;
-		String methodName = "";
-		String paramName = "";
-		boolean result;
+    @Test
+    void testIsValidParam() throws Exception {
+        DistributionEngine testSubject;
+        String paramValue = "";
+        String methodName = "";
+        String paramName = "";
+        boolean result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = Deencapsulation.invoke(testSubject, "isValidParam",
+                new Object[]{paramValue, methodName, paramName});
+    }
 
-		// test 1
-		testSubject = createTestSubject();
-		paramValue = null;
-		result = Deencapsulation.invoke(testSubject, "isValidObject",
-				new Object[] { Object.class, methodName, paramName });
-		Assert.assertEquals(false, result);
-	}
-
-	@Test
-	public void testGetEnvironmentErrorDescription() throws Exception {
-		DistributionEngine testSubject;
-		StorageOperationStatus status = null;
-		String result;
+    @Test
+    void testIsValidParam_1() throws Exception {
+        DistributionEngine testSubject;
+        List<String> paramValue = null;
+        String methodName = "";
+        String paramName = "";
+        boolean result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = Deencapsulation.invoke(testSubject, "isValidParam",
+                new Object[]{List.class, methodName, paramName});
+    }
 
-		// default test
-		testSubject = createTestSubject();
-		result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
-				 StorageOperationStatus.DISTR_ENVIRONMENT_NOT_AVAILABLE);
-		result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
-				 StorageOperationStatus.DISTR_ENVIRONMENT_NOT_FOUND);
-		result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
-				 StorageOperationStatus.DISTR_ENVIRONMENT_SENT_IS_INVALID);
-		result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
-				 StorageOperationStatus.ARTIFACT_NOT_FOUND);
-	}
-
-	@Test
-	public void testIsEnvironmentAvailable() throws Exception {
-		DistributionEngine testSubject;
-		String envName = "";
-		StorageOperationStatus result;
+    @Test
+    void testIsValidObject() throws Exception {
+        DistributionEngine testSubject;
+        Object paramValue = null;
+        String methodName = "";
+        String paramName = "";
+        boolean result;
+
+        // test 1
+        testSubject = createTestSubject();
+        paramValue = null;
+        result = Deencapsulation.invoke(testSubject, "isValidObject",
+                new Object[]{Object.class, methodName, paramName});
+        assertEquals(false, result);
+    }
 
-		// test 1
-		testSubject = createTestSubject();
-		envName = null;
-		result = testSubject.isEnvironmentAvailable(envName);
-		Assert.assertEquals(StorageOperationStatus.DISTR_ENVIRONMENT_SENT_IS_INVALID, result);
+    @Test
+    void testGetEnvironmentErrorDescription() throws Exception {
+        DistributionEngine testSubject;
+        StorageOperationStatus status = null;
+        String result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
+                StorageOperationStatus.DISTR_ENVIRONMENT_NOT_AVAILABLE);
+        result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
+                StorageOperationStatus.DISTR_ENVIRONMENT_NOT_FOUND);
+        result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
+                StorageOperationStatus.DISTR_ENVIRONMENT_SENT_IS_INVALID);
+        result = Deencapsulation.invoke(testSubject, "getEnvironmentErrorDescription",
+                StorageOperationStatus.ARTIFACT_NOT_FOUND);
+    }
 
-		// test 2
-		testSubject = createTestSubject();
-		envName = "mock";
-		result = testSubject.isEnvironmentAvailable(envName);
-		Assert.assertEquals(StorageOperationStatus.DISTR_ENVIRONMENT_NOT_FOUND, result);
-	}
+    @Test
+    void testIsEnvironmentAvailable() throws Exception {
+        DistributionEngine testSubject;
+        String envName = "";
+        StorageOperationStatus result;
+
+        // test 1
+        testSubject = createTestSubject();
+        envName = null;
+        result = testSubject.isEnvironmentAvailable(envName);
+        assertEquals(StorageOperationStatus.DISTR_ENVIRONMENT_SENT_IS_INVALID, result);
+
+        // test 2
+        testSubject = createTestSubject();
+        envName = "mock";
+        result = testSubject.isEnvironmentAvailable(envName);
+        assertEquals(StorageOperationStatus.DISTR_ENVIRONMENT_NOT_FOUND, result);
+    }
 
-	//TODO Create test coverage for this method
+    //TODO Create test coverage for this method
 	/*@Test
 	public void testIsEnvironmentAvailable_1() throws Exception {
 		DistributionEngine testSubject;
@@ -288,61 +291,61 @@ public class DistributionEngineTest{
 		result = testInstance.isEnvironmentAvailable();
 	}*/
 
-	@Test(expected=NullPointerException.class)
-	public void testDisableEnvironment() throws Exception {
-		DistributionEngine testSubject;
-		String envName = "";
-
-		// default test
-		testSubject = createTestSubject();
-		testSubject.disableEnvironment(envName);
-	}
-
-	@Test
-	public void testBuildTopicName() throws Exception {
-		DistributionEngine testSubject;
-		String envName = "";
-		String result;
-
-		// default test
-		testSubject = createTestSubject();
-		result = Deencapsulation.invoke(testSubject, "buildTopicName", new Object[] { envName });
-	}
-
-	@Test
-	public void testIsReadyForDistribution() throws Exception {
-		DistributionEngine testSubject;
-		Service service = null;
-		String envName = "";
-		StorageOperationStatus result;
+    @Test
+    void testDisableEnvironment() throws Exception {
+        DistributionEngine testSubject;
+        String envName = "";
 
-		// default test
-		testSubject = createTestSubject();
-		result = testSubject.isReadyForDistribution(envName);
-	}
+        // default test
+        testSubject = createTestSubject();
+        assertThrows(NullPointerException.class, () -> testSubject.disableEnvironment(envName));
+    }
 
+    @Test
+    void testBuildTopicName() throws Exception {
+        DistributionEngine testSubject;
+        String envName = "";
+        String result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = Deencapsulation.invoke(testSubject, "buildTopicName", new Object[]{envName});
+    }
 
-	@Test
-	public void testGetEnvironmentById() throws Exception {
-		DistributionEngine testSubject;
-		String opEnvId = "";
-		OperationalEnvironmentEntry result;
+    @Test
+    void testIsReadyForDistribution() throws Exception {
+        DistributionEngine testSubject;
+        Service service = null;
+        String envName = "";
+        StorageOperationStatus result;
+
+        // default test
+        testSubject = createTestSubject();
+        result = testSubject.isReadyForDistribution(envName);
+    }
 
-		// default test
-		when(environmentsEngine.getEnvironmentById(ArgumentMatchers.anyString())).thenReturn(new OperationalEnvironmentEntry());
-		result = testInstance.getEnvironmentById(opEnvId);
-	}
 
-	@Test
-	public void testBuildServiceForDistribution() throws Exception {
-		Service service = new Service();
-		String distributionId = "";
-		String workloadContext = "";
-		INotificationData result;
+    @Test
+    void testGetEnvironmentById() throws Exception {
+        DistributionEngine testSubject;
+        String opEnvId = "";
+        OperationalEnvironmentEntry result;
+
+        // default test
+        when(environmentsEngine.getEnvironmentById(ArgumentMatchers.anyString())).thenReturn(new OperationalEnvironmentEntry());
+        result = testInstance.getEnvironmentById(opEnvId);
+    }
 
-		// default test
-		//testSubject = createTestSubject();
-		when(serviceDistributionArtifactsBuilder.buildResourceInstanceForDistribution(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new NotificationDataImpl());
-		result = testInstance.buildServiceForDistribution(service, distributionId, workloadContext);
-	}
+    @Test
+    void testBuildServiceForDistribution() throws Exception {
+        Service service = new Service();
+        String distributionId = "";
+        String workloadContext = "";
+        INotificationData result;
+
+        // default test
+        //testSubject = createTestSubject();
+        when(serviceDistributionArtifactsBuilder.buildResourceInstanceForDistribution(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())).thenReturn(new NotificationDataImpl());
+        result = testInstance.buildServiceForDistribution(service, distributionId, workloadContext);
+    }
 }
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogicBaseTestSetup.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogicBaseTestSetup.java
index cc6f676618..049f8a5541 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogicBaseTestSetup.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/impl/ServiceBusinessLogicBaseTestSetup.java
@@ -204,7 +204,7 @@ class ServiceBusinessLogicBaseTestSetup extends BaseBusinessLogicMock {
             interfaceLifecycleTypeOperation, artifactBl, distributionEngine, componentInstanceBusinessLogic, serviceDistributionValidation,
             forwardingPathValidator, uiComponentDataConverter, artifactToscaOperation, componentContactIdValidator, componentNameValidator,
             componentTagsValidator, componentValidator, componentIconValidator, componentProjectCodeValidator, componentDescriptionValidator,
-            modelOperation, serviceRoleValidator, serviceInstantiationTypeValidator, serviceCategoryValidator, serviceValidator);
+            modelOperation, serviceRoleValidator, serviceInstantiationTypeValidator, serviceCategoryValidator, serviceValidator, null);
         bl.setComponentContactIdValidator(componentContactIdValidator);
         bl.setComponentIconValidator(componentIconValidator);
         bl.setComponentTagsValidator(componentTagsValidator);
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/kafka/KafkaHandlerTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/kafka/KafkaHandlerTest.java
index 91ee0235ad..ecdd75092b 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/kafka/KafkaHandlerTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/kafka/KafkaHandlerTest.java
@@ -19,33 +19,31 @@
  */
 package org.openecomp.sdc.be.components.kafka;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.when;
-
 import com.google.gson.JsonSyntaxException;
+import fj.data.Either;
 import org.apache.kafka.common.KafkaException;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.junit.jupiter.MockitoExtension;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
-
-import java.util.ArrayList;
-import fj.data.Either;
-import java.util.List;
-
+import org.mockito.junit.jupiter.MockitoExtension;
 import org.openecomp.sdc.be.components.distribution.engine.CambriaErrorResponse;
-import org.openecomp.sdc.be.components.distribution.engine.NotificationDataImpl;
 import org.openecomp.sdc.be.components.distribution.engine.INotificationData;
+import org.openecomp.sdc.be.components.distribution.engine.NotificationDataImpl;
 import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
-public class KafkaHandlerTest {
+class KafkaHandlerTest {
 
     @Mock
     private SdcKafkaConsumer mockSdcKafkaConsumer;
@@ -55,26 +53,28 @@ public class KafkaHandlerTest {
 
     private KafkaHandler kafkaHandler;
 
+    @BeforeEach
+    void setup() {
+        kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
+    }
+
     @Test
-    public void testIsKafkaActiveTrue(){
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
+    void testIsKafkaActiveTrue() {
         assertTrue(kafkaHandler.isKafkaActive());
     }
 
     @Test
-    public void testIsKafkaActiveFalse(){
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
+    void testIsKafkaActiveFalse() {
         kafkaHandler.setKafkaActive(false);
         assertFalse(kafkaHandler.isKafkaActive());
     }
 
     @Test
-    public void testFetchFromTopicSuccess(){
+    void testFetchFromTopicSuccess() {
         String testTopic = "testTopic";
         List<String> mockedReturnedMessages = new ArrayList<>();
         mockedReturnedMessages.add("message1");
         mockedReturnedMessages.add("message2");
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         when(mockSdcKafkaConsumer.poll(any())).thenReturn(mockedReturnedMessages);
         Either<Iterable<String>, CambriaErrorResponse> response = kafkaHandler.fetchFromTopic(testTopic);
         Iterable<String> actualReturnedMessages = response.left().value();
@@ -83,9 +83,8 @@ public class KafkaHandlerTest {
     }
 
     @Test
-    public void testFetchFromTopicFail(){
+    void testFetchFromTopicFail() {
         String testTopic = "testTopic";
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         when(mockSdcKafkaConsumer.poll(any())).thenThrow(new KafkaException());
         Either<Iterable<String>, CambriaErrorResponse> response = kafkaHandler.fetchFromTopic(testTopic);
         CambriaErrorResponse responseValue = response.right().value();
@@ -94,9 +93,8 @@ public class KafkaHandlerTest {
     }
 
     @Test
-    public void testSendNotificationSuccess(){
+    void testSendNotificationSuccess() {
         String testTopic = "testTopic";
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         INotificationData testData = new NotificationDataImpl();
         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
         assertEquals(response.getOperationStatus(), CambriaOperationStatus.OK);
@@ -104,9 +102,8 @@ public class KafkaHandlerTest {
     }
 
     @Test
-    public void testSendNotificationKafkaException(){
+    void testSendNotificationKafkaException() {
         String testTopic = "testTopic";
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         INotificationData testData = new NotificationDataImpl();
         doThrow(KafkaException.class).when(mockSdcKafkaProducer).send(any(), any());
         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
@@ -115,9 +112,8 @@ public class KafkaHandlerTest {
     }
 
     @Test
-    public void testSendNotificationJsonSyntaxException(){
+    void testSendNotificationJsonSyntaxException() {
         String testTopic = "testTopic";
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         INotificationData testData = new NotificationDataImpl();
         doThrow(JsonSyntaxException.class).when(mockSdcKafkaProducer).send(any(), any());
         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
@@ -126,9 +122,8 @@ public class KafkaHandlerTest {
     }
 
     @Test
-    public void testSendNotificationFlushException(){
+    void testSendNotificationFlushException() {
         String testTopic = "testTopic";
-        KafkaHandler kafkaHandler = new KafkaHandler(mockSdcKafkaConsumer, mockSdcKafkaProducer, true);
         INotificationData testData = new NotificationDataImpl();
         doThrow(KafkaException.class).when(mockSdcKafkaProducer).flush();
         CambriaErrorResponse response = kafkaHandler.sendNotification(testTopic, testData);
diff --git a/catalog-be/src/test/java/org/openecomp/sdc/be/components/path/ForwardingPathBusinessLogicTest.java b/catalog-be/src/test/java/org/openecomp/sdc/be/components/path/ForwardingPathBusinessLogicTest.java
index 926522ecdc..2b4e90ffe9 100644
--- a/catalog-be/src/test/java/org/openecomp/sdc/be/components/path/ForwardingPathBusinessLogicTest.java
+++ b/catalog-be/src/test/java/org/openecomp/sdc/be/components/path/ForwardingPathBusinessLogicTest.java
@@ -43,16 +43,16 @@ import org.openecomp.sdc.exception.ResponseFormat;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 @SpringJUnitConfig(locations = "classpath:paths/path-context.xml")
-public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
+class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
 
     @BeforeAll
     static void setup() {
         configurationManager =
-            new ConfigurationManager(new FSConfigurationSource(ExternalConfiguration.getChangeListener(), "src/test/resources/config/catalog-be"));
+                new ConfigurationManager(new FSConfigurationSource(ExternalConfiguration.getChangeListener(), "src/test/resources/config/catalog-be"));
     }
 
     @Test
-    public void shouldFailToUpdateForwardingPathSincePathDoesNotExist() {
+    void shouldFailToUpdateForwardingPathSincePathDoesNotExist() {
         Assertions.assertThrows(ComponentException.class, () -> {
             Service service = initForwardPath();
             bl.updateForwardingPath(FORWARDING_PATH_ID, service, user, true);
@@ -60,7 +60,7 @@ public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
     }
 
     @Test
-    public void shouldFailToDeleteForwardingPathSincePathDoesNotExist() {
+    void shouldFailToDeleteForwardingPathSincePathDoesNotExist() {
         Assertions.assertThrows(ComponentException.class, () -> {
             initForwardPath();
             bl.deleteForwardingPaths("delete_forward_test", Sets.newHashSet(FORWARDING_PATH_ID), user, true);
@@ -68,7 +68,7 @@ public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
     }
 
     @Test
-    public void shouldSucceedCreateAndDeleteForwardingPath() {
+    void shouldSucceedCreateAndDeleteForwardingPath() {
         Service createdService = createService();
         Service service = initForwardPath();
         assertNotNull(service);
@@ -81,7 +81,7 @@ public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
 
         // should return the created path
         Either<UiComponentDataTransfer, ResponseFormat> uiResaponse = bl.getComponentDataFilteredByParams(createdService.getUniqueId(), user,
-            Lists.newArrayList(ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
+                Lists.newArrayList(ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
         assertTrue(uiResaponse.isLeft());
         UiServiceDataTransfer uiServiceDataTransfer = (UiServiceDataTransfer) uiResaponse.left().value();
         Map<String, ForwardingPathDataDefinition> forwardingPaths = uiServiceDataTransfer.getForwardingPaths();
@@ -104,7 +104,7 @@ public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
 
         // make sure changes were applied
         uiResaponse = bl.getComponentDataFilteredByParams(createdService.getUniqueId(), user,
-            Lists.newArrayList(ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
+                Lists.newArrayList(ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
         assertTrue(uiResaponse.isLeft());
         uiServiceDataTransfer = (UiServiceDataTransfer) uiResaponse.left().value();
         Map<String, ForwardingPathDataDefinition> forwardingPathsUpdated = uiServiceDataTransfer.getForwardingPaths();
@@ -119,7 +119,7 @@ public class ForwardingPathBusinessLogicTest extends BaseForwardingPathTest {
 
         // nothing to return now
         uiResaponse = bl.getComponentDataFilteredByParams(createdService.getUniqueId(), user,
-            Lists.newArrayList(ComponentFieldsEnum.COMPONENT_INSTANCES.getValue(), ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
+                Lists.newArrayList(ComponentFieldsEnum.COMPONENT_INSTANCES.getValue(), ComponentFieldsEnum.FORWARDING_PATHS.getValue()));
         assertTrue(uiResaponse.isLeft());
         uiServiceDataTransfer = (UiServiceDataTransfer) uiResaponse.left().value();
         forwardingPaths = uiServiceDataTransfer.getForwardingPaths();
diff --git a/catalog-be/src/test/resources/paths/path-context.xml b/catalog-be/src/test/resources/paths/path-context.xml
index 53e8b8c74b..c32f32b365 100644
--- a/catalog-be/src/test/resources/paths/path-context.xml
+++ b/catalog-be/src/test/resources/paths/path-context.xml
@@ -21,49 +21,49 @@ Modifications copyright (c) 2018 Nokia
 ================================================================================
 -->
 <beans xmlns:aop="http://www.springframework.org/schema/aop"
-    xmlns:context="http://www.springframework.org/schema/context"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xmlns="http://www.springframework.org/schema/beans"
-    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
+       xmlns:context="http://www.springframework.org/schema/context"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns="http://www.springframework.org/schema/beans"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
         http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd
         http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd">
 
     <aop:aspectj-autoproxy proxy-target-class="true"/>
     <bean class="org.openecomp.sdc.be.dao.config.DAOSpringConfig"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.ForwardingPathToscaOperationFacade"
-        id="tosca-operation-facade"/>
+          id="tosca-operation-facade"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.InMemoryJanusGraphClient"
-        id="janusgraph-client"/>
+          id="janusgraph-client"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.ArtifactCassandraDaoMock"
-        id="artifact-cassandra-dao"/>
+          id="artifact-cassandra-dao"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.AuditCassandraDaoMock"
-        id="audit-cassandra-dao"/>
+          id="audit-cassandra-dao"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.FeatureToggleDaoMock"
-        id="feature_toggle_dao"/>
+          id="feature_toggle_dao"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.EnvironmentCassandraDaoMock"
-        id="operational-environment-dao"/>
+          id="operational-environment-dao"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.SdcSchemaFilesCassandraDaoMock"
-        id="sdc-schema-files-cassandra-dao"/>
+          id="sdc-schema-files-cassandra-dao"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.ToscaModelImportCassandraDaoMock"
-        id="tosca-model-import-cassandra-dao"/>
+          id="tosca-model-import-cassandra-dao"/>
     <bean class="org.openecomp.sdc.be.client.onboarding.impl.OnboardingClientImpl"
-        id="onboarding-client"/>
+          id="onboarding-client"/>
     <bean class="org.openecomp.sdc.be.catalog.impl.DmaapProducer" id="dmaapProducer"/>
     <bean class="org.openecomp.sdc.be.catalog.impl.DmaapProducerHealth" id="dmaapProducerHealth"/>
     <bean class="org.openecomp.sdc.be.components.lifecycle.LifecycleBusinessLogic"
-        id="lifecycleBusinessLogic"/>
+          id="lifecycleBusinessLogic"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.DistributionEngineMock"
-        id="distributionEngine"/>
+          id="distributionEngine"/>
     <bean class="org.openecomp.sdc.be.components.path.beans.ForwardingPathValidatorMock"
-        id="forwardingPathValidator"/>
+          id="forwardingPathValidator"/>
     <bean class="org.openecomp.sdc.be.components.attribute.AttributeDeclarationOrchestrator"
-        id="attributeDeclarationOrchestrator"/>
-    <bean
-        class="org.openecomp.sdc.be.components.attribute.ComponentInstanceOutputAttributeDeclarator"/>
+          id="attributeDeclarationOrchestrator"/>
+    <bean class="org.openecomp.sdc.be.components.attribute.ComponentInstanceOutputAttributeDeclarator"/>
     <bean class="org.openecomp.sdc.be.components.attribute.ComponentInstanceAttributeDeclarator"/>
+    <bean class="org.openecomp.sdc.be.components.kafka.KafkaHandler"/>
     <context:annotation-config/>
     <context:component-scan
-        base-package="
+            base-package="
         org.openecomp.sdc.be.dao.janusgraph,
         org.openecomp.sdc.be.model.jsonjanusgraph,
    		org.openecomp.sdc.be.model.operations.impl,
@@ -80,6 +80,7 @@ Modifications copyright (c) 2018 Nokia
 		org.openecomp.sdc.be.components.csar,
 	  	org.openecomp.sdc.be.impl,
         org.openecomp.sdc.be.tosca,
+        org.openecomp.sdc.be.components.kafka,
         org.openecomp.sdc.be.dao.impl">
     </context:component-scan>