Config fetch for VESCollector through DCAE-SDK (CBS Client)

Issue-ID: DCAEGEN2-2212
Signed-off-by: Pawel <pawel.kasperkiewicz@nokia.com>
Change-Id: I25072b340b5c9f2f538d39e5befb1331804b7bba
Signed-off-by: Pawel <pawel.kasperkiewicz@nokia.com>

35 files changed, 857 insertions, 973 deletions

diff --git a/Changelog.md b/Changelog.md
index 51211b43..73291b42 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -24,4 +24,6 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
           Fix error response model
           Update DPO model
 ## [1.7.3] - 10/08/2020
-        - [DCAEGEN2-2264](https://jira.onap.org/browse/DCAEGEN2-2264) - Add implementation of stndDefined fields validation
\ No newline at end of file
+        - [DCAEGEN2-2264](https://jira.onap.org/browse/DCAEGEN2-2264) - Add implementation of stndDefined fields validation
+## [1.7.4] - 04/08/2020
+        - [DCAEGEN2-2212](https://jira.onap.org/browse/DCAEGEN2-2212) - Config fetch for VESCollector through DCAE-SDK (CBS Client)        
diff --git a/README.md b/README.md
index 0d41ac8a..7b93a989 100644
--- a/README.md
+++ b/README.md
@@ -64,12 +64,9 @@ a particular deployment system.
 Variables set manually / coming from deployment system:
 - COLLECTOR_IP
 - DMAAPHOST - should contain an address to DMaaP, so that event publishing can work
-- CBSPOLLTIMER - it should be put in here if we want to automatically fetch configuration from CBS.
-- CONSUL_PROTOCOL - Consul protocol by default set to **http**, if it is need to change it then that can be set to different value 
-- CONSUL_HOST - used with conjunction with CBSPOLLTIMER, should be a host address (without port! e.g my-ip-or-host) where Consul service lies
-- CBS_PROTOCOL - Config Binding Service protocol by default set to **http**, if it is need to change it then that can be set to different value
-- CONFIG_BINDING_SERVICE - used with conjunction with CBSPOLLTIMER, should be a name of CBS as it is registered in Consul
-- HOSTNAME - used with conjunction with CBSPOLLTIMER, should be a name of VESCollector application as it is registered in CBS catalog
+- CONFIG_BINDING_SERVICE - should be a name of CBS
+- CONFIG_BINDING_SERVICE_SERVICE_PORT - should be an http port of CBS
+- HOSTNAME - should be a name of VESCollector application as it is registered in CBS catalog
 
 ### Docker file system layout
 The main directory where all code resides in docker container
@@ -113,9 +110,11 @@ For testing purpose, the docker image includes preset configuration which can be
 ### Consul - Dynamic configuration 
 
 
-Application properties like /etc/collector.properties and Dmaap configuration /etc/DmaapConfig.json are updated frequently by configuration stored in Consul(CBS) http://<kubernetes_host_ip>:30270/ui/#/dc1/kv/<vescollector_SCN> 
+Application properties like /etc/collector.properties and Dmaap configuration /etc/DmaapConfig.json are updated frequently by configuration stored in Consul(CBS)
+http://<kubernetes_host_ip>:30270/ui/#/dc1/kv/<vescollector_SCN> 
 Configuration stored in Consul have bigger priority and always will override local configuration so all configuration modification should be done using Consul update on corresponding kv store. 
 Frequently how often configuration will be fetch from Consul server is manageable in /etc/collector.properties property "collector.dynamic.config.update.frequency={time in minutes}".
+To fetch configuration from Consul, VES collector uses CBS client from DCAE SDK.
 
 Sample configuration of VESCollector K-V store can be found under /dpo/data-formats/ConsulConfig.json
 
diff --git a/pom.xml b/pom.xml
index 3ba5e35a..561eca6b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -24,7 +24,7 @@
   </parent>

   <groupId>org.onap.dcaegen2.collectors.ves</groupId>

   <artifactId>VESCollector</artifactId>

-  <version>1.7.3-SNAPSHOT</version>

+  <version>1.7.4-SNAPSHOT</version>

   <name>dcaegen2-collectors-ves</name>

   <description>VESCollector</description>

   <properties>

@@ -79,6 +79,7 @@
     <api-custom-header.version>1.1.4</api-custom-header.version>

     <functionaljava.version>4.8.1</functionaljava.version>

     <external-schema-manager.version>1.4.1</external-schema-manager.version>

+    <sdk.version>1.4.2</sdk.version>

   </properties>

   <build>

     <pluginManagement>

@@ -461,6 +462,11 @@
       <version>${functionaljava.version}</version>

       <scope>compile</scope>

     </dependency>

+    <dependency>

+      <groupId>org.onap.dcaegen2.services.sdk.rest.services</groupId>

+      <artifactId>cbs-client</artifactId>

+      <version>${sdk.version}</version>

+    </dependency>

   </dependencies>

   <repositories>

     <repository>

diff --git a/src/main/java/org/onap/dcae/VesApplication.java b/src/main/java/org/onap/dcae/VesApplication.java
index bb785dbd..e5ee6e35 100644
--- a/src/main/java/org/onap/dcae/VesApplication.java
+++ b/src/main/java/org/onap/dcae/VesApplication.java
@@ -22,16 +22,13 @@
 package org.onap.dcae;
 
 import io.vavr.collection.Map;
-import java.nio.file.Paths;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 import org.onap.dcae.common.EventSender;
 import org.onap.dcae.common.validator.StndDefinedValidatorResolver;
 import org.onap.dcae.common.publishing.DMaaPConfigurationParser;
-import org.onap.dcae.common.publishing.EventPublisher;
+import org.onap.dcae.common.publishing.DMaaPEventPublisher;
 import org.onap.dcae.common.publishing.PublisherConfig;
-import org.onap.dcae.controller.ConfigLoader;
+import org.onap.dcae.configuration.ConfigLoader;
+import org.onap.dcae.configuration.ConfigLoaderFactory;
 import org.onap.dcaegen2.services.sdk.services.external.schema.manager.service.StndDefinedValidator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -44,74 +41,71 @@ import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Lazy;
 
+import java.nio.file.Paths;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 @SpringBootApplication(exclude = {GsonAutoConfiguration.class, SecurityAutoConfiguration.class})
 public class VesApplication {
 
     private static final Logger incomingRequestsLogger = LoggerFactory.getLogger("org.onap.dcae.common.input");
-    private static final Logger oplog = LoggerFactory.getLogger("org.onap.dcae.common.output");
     private static final Logger errorLog = LoggerFactory.getLogger("org.onap.dcae.common.error");
     private static ApplicationSettings applicationSettings;
     private static ConfigurableApplicationContext context;
     private static ConfigLoader configLoader;
     private static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
-    private static SpringApplication app;
-    private static EventPublisher eventPublisher;
+    private static DMaaPEventPublisher eventPublisher;
     private static ScheduledFuture<?> scheduleFeatures;
 
     public static void main(String[] args) {
-      app = new SpringApplication(VesApplication.class);
-      applicationSettings = new ApplicationSettings(args, CLIUtils::processCmdLine);
-      scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
-      init();
-      app.setAddCommandLineProperties(true);
-      context = app.run();
-      configLoader.updateConfig();
+        SpringApplication app = new SpringApplication(VesApplication.class);
+        applicationSettings = new ApplicationSettings(args, CLIUtils::processCmdLine);
+        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
+        init();
+        app.setAddCommandLineProperties(true);
+        context = app.run();
+        configLoader.updateConfig();
     }
 
     public static void restartApplication() {
-      Thread thread = new Thread(() -> {
-        context.close();
-        applicationSettings.reloadProperties();
-        scheduleFeatures.cancel(true);
-        init();
-        context = SpringApplication.run(VesApplication.class);
-      });
-      thread.setDaemon(false);
-      thread.start();
+        Thread thread = new Thread(() -> {
+            context.close();
+            applicationSettings.reloadProperties();
+            scheduleFeatures.cancel(true);
+            init();
+            context = SpringApplication.run(VesApplication.class);
+        });
+        thread.setDaemon(false);
+        thread.start();
     }
 
     private static void init() {
-      createConfigLoader();
-      createSchedulePoolExecutor();
-      createExecutors();
+        createConfigLoader();
+        createSchedulePoolExecutor();
+        createExecutors();
     }
 
     private static void createExecutors() {
-      eventPublisher = EventPublisher.createPublisher(oplog, getDmapConfig());
+        eventPublisher = new DMaaPEventPublisher(getDmaapConfig());
     }
 
     private static void createSchedulePoolExecutor() {
-      scheduleFeatures = scheduledThreadPoolExecutor.scheduleAtFixedRate(configLoader::updateConfig,
-          applicationSettings.configurationUpdateFrequency(),
-          applicationSettings.configurationUpdateFrequency(),
-          TimeUnit.MINUTES);
+        scheduleFeatures = scheduledThreadPoolExecutor.scheduleAtFixedRate(configLoader::updateConfig,
+                applicationSettings.configurationUpdateFrequency(),
+                applicationSettings.configurationUpdateFrequency(),
+                TimeUnit.MINUTES);
     }
 
     private static void createConfigLoader() {
-      configLoader = ConfigLoader.create(getEventPublisher()::reconfigure,
-          Paths.get(applicationSettings.dMaaPConfigurationFileLocation()),
-          applicationSettings.configurationFileLocation());
-    }
-
-
-    private static EventPublisher getEventPublisher() {
-      return EventPublisher.createPublisher(oplog, DMaaPConfigurationParser
-          .parseToDomainMapping(Paths.get(applicationSettings.dMaaPConfigurationFileLocation())).get());
+        configLoader = ConfigLoaderFactory.create(
+                applicationSettings.configurationFileLocation(),
+                Paths.get(applicationSettings.dMaaPConfigurationFileLocation()));
     }
 
-    private static Map<String, PublisherConfig> getDmapConfig() {
-      return DMaaPConfigurationParser
-          .parseToDomainMapping(Paths.get(applicationSettings.dMaaPConfigurationFileLocation())).get();
+    private static Map<String, PublisherConfig> getDmaapConfig() {
+        return DMaaPConfigurationParser
+                .parseToDomainMapping(Paths.get(applicationSettings.dMaaPConfigurationFileLocation())).get();
     }
 
     @Bean
diff --git a/src/main/java/org/onap/dcae/common/EventSender.java b/src/main/java/org/onap/dcae/common/EventSender.java
index 63be9106..1ec918c9 100644
--- a/src/main/java/org/onap/dcae/common/EventSender.java
+++ b/src/main/java/org/onap/dcae/common/EventSender.java
@@ -3,7 +3,7 @@
  * VES Collector
  * ================================================================================
  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2020 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -25,7 +25,7 @@ import com.att.nsa.logging.LoggingContext;
 import com.att.nsa.logging.log4j.EcompFields;
 import io.vavr.collection.Map;
 import org.onap.dcae.common.model.VesEvent;
-import org.onap.dcae.common.publishing.EventPublisher;
+import org.onap.dcae.common.publishing.DMaaPEventPublisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,10 +35,10 @@ public class EventSender {
 
   private static final Logger metriclog = LoggerFactory.getLogger("com.att.ecomp.metrics");
   private Map<String, String[]> streamIdToDmaapIds;
-  private EventPublisher eventPublisher;
+  private DMaaPEventPublisher eventPublisher;
   private static final Logger log = LoggerFactory.getLogger(EventSender.class);
 
-  public EventSender(EventPublisher eventPublisher, Map<String, String[]> streamIdToDmaapIds) {
+  public EventSender(DMaaPEventPublisher eventPublisher, Map<String, String[]> streamIdToDmaapIds) {
     this.eventPublisher = eventPublisher;
     this.streamIdToDmaapIds = streamIdToDmaapIds;
   }
diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
index 3fc9e254..4e9aabc7 100644
--- a/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
+++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
@@ -3,7 +3,7 @@
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
  * Copyright (C) 2017,2020 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -39,7 +39,7 @@ import static org.onap.dcae.common.publishing.VavrUtils.f;
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
  */
-class DMaaPEventPublisher implements EventPublisher {
+public class DMaaPEventPublisher {
     private static final int PENDING_MESSAGE_LOG_THRESHOLD = 100;
     private static final String VES_UNIQUE_ID = "VESuniqueId";
     private static final String EVENT = "event";
@@ -47,15 +47,16 @@ class DMaaPEventPublisher implements EventPublisher {
     private static final String PARTITION_KEY = "sourceName";
     private static final Logger log = LoggerFactory.getLogger(DMaaPEventPublisher.class);
     private final DMaaPPublishersCache publishersCache;
-    private final Logger outputLogger;
+    private final Logger outputLogger = LoggerFactory.getLogger("org.onap.dcae.common.output");
 
-    DMaaPEventPublisher(DMaaPPublishersCache publishersCache,
-                        Logger outputLogger) {
+    DMaaPEventPublisher(DMaaPPublishersCache publishersCache) {
         this.publishersCache = publishersCache;
-        this.outputLogger = outputLogger;
     }
 
-    @Override
+    public DMaaPEventPublisher(Map<String, PublisherConfig> dMaaPConfig) {
+        this(new DMaaPPublishersCache(dMaaPConfig));
+    }
+
     public void sendEvent(JSONObject event, String domain) {
         clearVesUniqueIdFromEvent(event);
         publishersCache.getPublisher(domain)
@@ -64,11 +65,6 @@ class DMaaPEventPublisher implements EventPublisher {
                 .forEach(publisher -> sendEvent(event, domain, publisher));
     }
 
-    @Override
-    public void reconfigure(Map<String, PublisherConfig> dMaaPConfig) {
-        publishersCache.reconfigure(dMaaPConfig);
-    }
-
     private void sendEvent(JSONObject event, String domain, CambriaBatchingPublisher publisher) {
         Try.run(() -> uncheckedSendEvent(event, domain, publisher))
                 .onFailure(exc -> closePublisher(event, domain, exc));
diff --git a/src/main/java/org/onap/dcae/controller/ConfigFilesFacade.java b/src/main/java/org/onap/dcae/configuration/ConfigFilesFacade.java
index 0b2c197d..c0280c75 100644
--- a/src/main/java/org/onap/dcae/controller/ConfigFilesFacade.java
+++ b/src/main/java/org/onap/dcae/configuration/ConfigFilesFacade.java
@@ -3,7 +3,7 @@
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2020 Nokia. 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.
@@ -18,37 +18,38 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.controller;
+package org.onap.dcae.configuration;
+
+import static io.vavr.API.Try;
+import static org.onap.dcae.common.publishing.VavrUtils.enhanceError;
+import static org.onap.dcae.common.publishing.VavrUtils.f;
+import static org.onap.dcae.common.publishing.VavrUtils.logError;
+import static org.onap.dcae.configuration.Conversions.toList;
 
 import io.vavr.CheckedRunnable;
 import io.vavr.Tuple2;
 import io.vavr.collection.Map;
 import io.vavr.control.Try;
+import java.io.FileNotFoundException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.configuration.PropertiesConfiguration;
 import org.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.FileNotFoundException;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-
-import static io.vavr.API.Try;
-import static org.onap.dcae.common.publishing.VavrUtils.*;
-import static org.onap.dcae.controller.Conversions.toList;
-
 class ConfigFilesFacade {
 
-    private static Logger log = LoggerFactory.getLogger(ConfigFilesFacade.class);
+    private static final Logger log = LoggerFactory.getLogger(ConfigFilesFacade.class);
 
     private final Path dMaaPConfigPath;
     private final Path propertiesPath;
 
-    public ConfigFilesFacade(Path dMaaPConfigPath, Path propertiesPath) {
-        this.dMaaPConfigPath = dMaaPConfigPath;
+    ConfigFilesFacade(Path propertiesPath, Path dMaaPConfigPath) {
         this.propertiesPath = propertiesPath;
+        this.dMaaPConfigPath = dMaaPConfigPath;
     }
 
     Try<Map<String, String>> readCollectorProperties() {
@@ -126,5 +127,4 @@ class ConfigFilesFacade {
     private String indentConfiguration(String configuration) {
         return new JSONObject(configuration).toString(4);
     }
-
-}
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/configuration/ConfigLoader.java b/src/main/java/org/onap/dcae/configuration/ConfigLoader.java
new file mode 100644
index 00000000..eaeb09e3
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/ConfigLoader.java
@@ -0,0 +1,99 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.onap.dcae.configuration.cbs.CbsConfigResolver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConfigLoader {
+
+    private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class);
+    private final ConfigFilesFacade configFilesFacade;
+    private final CbsConfigResolver cbsConfigResolver;
+    private final Runnable applicationRestarter;
+    private boolean isApplicationRestartNeeded;
+
+    ConfigLoader(ConfigFilesFacade configFilesFacade, CbsConfigResolver cbsConfigResolver, Runnable applicationRestarter) {
+        this.configFilesFacade = configFilesFacade;
+        this.cbsConfigResolver = cbsConfigResolver;
+        this.applicationRestarter = applicationRestarter;
+        this.isApplicationRestartNeeded = false;
+    }
+
+    public synchronized void updateConfig() {
+        Option<JSONObject> appConfig = cbsConfigResolver.getAppConfig();
+        appConfig.peek(this::handleUpdate).onEmpty(logSkipMessage());
+    }
+
+    private Runnable logSkipMessage() {
+        return () -> log.info("Skipping dynamic configuration");
+    }
+
+    private void handleUpdate(JSONObject appConfig) {
+        updatePropertiesIfChanged(appConfig);
+        updateDmaapConfigIfChanged(appConfig);
+        restartApplicationIfNeeded();
+    }
+
+    private void updatePropertiesIfChanged(JSONObject appConfig) {
+        Map<String, String> newProperties = ConfigParsing.getProperties(appConfig);
+        Map<String, String> oldProperties = configFilesFacade.readCollectorProperties().get();
+
+        if (!areCommonPropertiesSame(oldProperties, newProperties)) {
+            configFilesFacade.writeProperties(newProperties);
+            isApplicationRestartNeeded = true;
+        }
+    }
+
+    private boolean areCommonPropertiesSame(Map<String, String> oldProperties, Map<String, String> newProperties) {
+        Map<String, String> filteredOldProperties = filterIntersectingKeys(oldProperties, newProperties);
+        return filteredOldProperties.equals(newProperties);
+    }
+
+    private Map<String, String> filterIntersectingKeys(Map<String, String> primaryProperties,
+        Map<String, String> otherProperties) {
+        return primaryProperties.filterKeys(key -> containsKey(key, otherProperties));
+    }
+
+    private boolean containsKey(String key, Map<String, String> properties) {
+        return properties.keySet().contains(key);
+    }
+
+    private void updateDmaapConfigIfChanged(JSONObject appConfig) {
+        JSONObject oldDmaapConfig = configFilesFacade.readDMaaPConfiguration().get();
+        JSONObject newDmaapConfig = ConfigParsing.getDMaaPConfig(appConfig).get();
+
+        if (!oldDmaapConfig.similar(newDmaapConfig)) {
+            configFilesFacade.writeDMaaPConfiguration(newDmaapConfig);
+            isApplicationRestartNeeded = true;
+        }
+    }
+
+    private void restartApplicationIfNeeded() {
+        if (isApplicationRestartNeeded) {
+            applicationRestarter.run();
+            isApplicationRestartNeeded = false;
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/configuration/ConfigLoaderFactory.java b/src/main/java/org/onap/dcae/configuration/ConfigLoaderFactory.java
new file mode 100644
index 00000000..e5ea3c59
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/ConfigLoaderFactory.java
@@ -0,0 +1,37 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration;
+
+import java.nio.file.Path;
+import org.onap.dcae.VesApplication;
+import org.onap.dcae.configuration.cbs.CbsConfigResolver;
+import org.onap.dcae.configuration.cbs.CbsConfigResolverFactory;
+
+public class ConfigLoaderFactory {
+
+    public static ConfigLoader create(Path propertiesFile, Path dmaapConfigFile) {
+        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(propertiesFile, dmaapConfigFile);
+        CbsConfigResolver cbsConfigResolver = new CbsConfigResolverFactory().create();
+        return new ConfigLoader(
+            configFilesFacade,
+            cbsConfigResolver,
+            VesApplication::restartApplication);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/controller/ConfigParsing.java b/src/main/java/org/onap/dcae/configuration/ConfigParsing.java
index e1644222..13deff73 100644
--- a/src/main/java/org/onap/dcae/controller/ConfigParsing.java
+++ b/src/main/java/org/onap/dcae/configuration/ConfigParsing.java
@@ -3,7 +3,7 @@
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2020 Nokia. 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.
@@ -18,12 +18,12 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.controller;
+package org.onap.dcae.configuration;
 
 import static io.vavr.API.Try;
 import static io.vavr.API.Tuple;
 import static org.onap.dcae.common.publishing.VavrUtils.f;
-import static org.onap.dcae.controller.Conversions.toList;
+import static org.onap.dcae.configuration.Conversions.toList;
 
 import io.vavr.collection.Map;
 import io.vavr.control.Option;
@@ -54,5 +54,4 @@ interface ConfigParsing {
         log.info(f("Found following app properties: '%s'", confEntries));
         return confEntries;
     }
-
-}
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/controller/Conversions.java b/src/main/java/org/onap/dcae/configuration/Conversions.java
index e8f7cc0c..85c3f78d 100644
--- a/src/main/java/org/onap/dcae/controller/Conversions.java
+++ b/src/main/java/org/onap/dcae/configuration/Conversions.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -17,7 +17,7 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.controller;
+package org.onap.dcae.configuration;
 
 import static org.onap.dcae.common.publishing.VavrUtils.enhanceError;
 
@@ -28,7 +28,6 @@ import java.util.Iterator;
 import java.util.Spliterator;
 import java.util.Spliterators;
 import java.util.stream.StreamSupport;
-import org.json.JSONArray;
 import org.json.JSONObject;
 
 /**
@@ -41,13 +40,8 @@ interface Conversions {
             .mapFailure(enhanceError("Value '%s' is not a valid JSON document", strBody));
     }
 
-    static Try<JSONArray> toJsonArray(String strBody) {
-        return API.Try(() -> new JSONArray(strBody))
-            .mapFailure(enhanceError("Value '%s' is not a valid JSON array", strBody));
-    }
-
     static <T> List<T> toList(Iterator<T> iterator) {
-        return List.ofAll(StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false));
+        return List
+            .ofAll(StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false));
     }
-
-}
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolver.java b/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolver.java
new file mode 100644
index 00000000..360e0a8f
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolver.java
@@ -0,0 +1,61 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration.cbs;
+
+import io.vavr.control.Option;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class CbsClientConfigurationResolver {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(CbsClientConfigurationResolver.class);
+
+    private final String defaultProtocol = "http";
+    private final String defaultHostname = "config-binding-service";
+    private final int defaultPort = 10000;
+    private final String defaultAppName = "dcae-ves-collector";
+
+    CbsClientConfiguration resolveCbsClientConfiguration() {
+        try {
+            return CbsClientConfiguration.fromEnvironment();
+        } catch (Exception e) {
+            LOGGER.warn("Failed resolving CBS client configuration from system environments: " + e);
+        }
+        LOGGER.info("Falling back to use default CBS client configuration properties");
+        return getFallbackConfiguration();
+    }
+
+    private ImmutableCbsClientConfiguration getFallbackConfiguration() {
+        LOGGER.info("Falling back to use default CBS client configuration");
+        return createCbsClientConfiguration(defaultProtocol, defaultHostname, defaultAppName, defaultPort);
+    }
+
+    private ImmutableCbsClientConfiguration createCbsClientConfiguration(String protocol, String hostname,
+        String appName, Integer port) {
+        return ImmutableCbsClientConfiguration.builder()
+            .protocol(protocol)
+            .hostname(hostname)
+            .port(port)
+            .appName(appName)
+            .build();
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolver.java b/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolver.java
new file mode 100644
index 00000000..5a66c294
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolver.java
@@ -0,0 +1,58 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration.cbs;
+
+import com.google.gson.JsonObject;
+import io.vavr.control.Option;
+import java.util.function.Consumer;
+import org.json.JSONObject;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClientFactory;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsRequests;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsRequest;
+import org.onap.dcaegen2.services.sdk.rest.services.model.logging.RequestDiagnosticContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+
+public class CbsConfigResolver {
+
+    private static final Logger log = LoggerFactory.getLogger(CbsConfigResolver.class);
+
+    private final CbsClientConfiguration cbsClientConfiguration;
+    private final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+    private final CbsRequest cbsConfigurationRequest = CbsRequests.getConfiguration(diagnosticContext);
+
+    CbsConfigResolver(CbsClientConfiguration cbsClientConfiguration) {
+        this.cbsClientConfiguration = cbsClientConfiguration;
+    }
+
+    public Option<JSONObject> getAppConfig() {
+        JsonObject emptyJson = new JsonObject();
+        JsonObject jsonObject = CbsClientFactory.createCbsClient(cbsClientConfiguration)
+            .flatMap(cbsClient -> cbsClient.get(cbsConfigurationRequest))
+            .doOnError(error -> log.warn("Failed to fetch configuration from CBS " + error.getMessage()))
+            .onErrorReturn(emptyJson)
+            .block();
+
+        return emptyJson.equals(jsonObject) ? Option.none() : Option.of(new JSONObject(jsonObject.toString()));
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolverFactory.java b/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolverFactory.java
new file mode 100644
index 00000000..aa3ee00f
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolverFactory.java
@@ -0,0 +1,31 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration.cbs;
+
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+
+public class CbsConfigResolverFactory {
+
+    public CbsConfigResolver create() {
+        CbsClientConfigurationResolver resolver = new CbsClientConfigurationResolver();
+        CbsClientConfiguration cbsClientConfiguration = resolver.resolveCbsClientConfiguration();
+        return new CbsConfigResolver(cbsClientConfiguration);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/controller/ConfigLoader.java b/src/main/java/org/onap/dcae/controller/ConfigLoader.java
deleted file mode 100644
index dbf52823..00000000
--- a/src/main/java/org/onap/dcae/controller/ConfigLoader.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.controller;
-
-import static org.onap.dcae.common.publishing.DMaaPConfigurationParser.parseToDomainMapping;
-import static org.onap.dcae.controller.ConfigParsing.getDMaaPConfig;
-import static org.onap.dcae.controller.ConfigParsing.getProperties;
-import static org.onap.dcae.controller.EnvPropertiesReader.readEnvProps;
-
-import io.vavr.Function0;
-import io.vavr.Function1;
-import io.vavr.collection.HashMap;
-import io.vavr.collection.Map;
-import io.vavr.control.Try;
-import java.nio.file.Path;
-import java.util.function.Consumer;
-import org.json.JSONObject;
-import org.onap.dcae.VesApplication;
-import org.onap.dcae.common.publishing.PublisherConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ConfigLoader {
-
-    private static final String SKIP_MSG = "Skipping dynamic configuration update";
-    private static Logger log = LoggerFactory.getLogger(ConfigLoader.class);
-    private final Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer;
-    private final ConfigFilesFacade configFilesFacade;
-    private final Function1<EnvProps, Try<JSONObject>> configurationSource;
-    private final Function0<Map<String, String>> envVariablesSupplier;
-    private boolean toRestart = false;
-
-    ConfigLoader(Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer,
-        ConfigFilesFacade configFilesFacade,
-        Function1<EnvProps, Try<JSONObject>> configurationSource,
-        Function0<Map<String, String>> envVariablesSupplier) {
-        this.eventPublisherReconfigurer = eventPublisherReconfigurer;
-        this.configFilesFacade = configFilesFacade;
-        this.configurationSource = configurationSource;
-        this.envVariablesSupplier = envVariablesSupplier;
-    }
-
-    public static ConfigLoader create(
-        Consumer<Map<String, PublisherConfig>> eventPublisherReconfigurer,
-        Path dMaaPConfigFile, Path propertiesConfigFile) {
-        return new ConfigLoader(eventPublisherReconfigurer,
-            new ConfigFilesFacade(dMaaPConfigFile, propertiesConfigFile),
-            ConfigSource::getAppConfig,
-            () -> HashMap.ofAll(System.getenv()));
-    }
-
-    public void updateConfig() {
-        log.info("Trying to dynamically update config from Config Binding Service");
-        readEnvProps(envVariablesSupplier.get())
-            .onEmpty(() -> log.warn(SKIP_MSG)).forEach(this::updateConfig);
-    }
-
-    private void updateConfig(EnvProps props) {
-        configurationSource.apply(props)
-            .onFailure(logSkip())
-            .onSuccess(newConf -> {
-                        updateConfigurationProperties(newConf);
-                        updateDMaaPProperties(newConf);
-                        reloadApplication();
-                }
-            );
-    }
-
-    private void reloadApplication() {
-        if(toRestart){
-            log.info("New app config - Application will be restarted");
-            VesApplication.restartApplication();
-        }
-    }
-
-    private void updateDMaaPProperties(JSONObject newConf) {
-        configFilesFacade.readDMaaPConfiguration()
-            .onFailure(logSkip())
-            .onSuccess(oldDMaaPConf -> getDMaaPConfig(newConf)
-                .onEmpty(() -> log.warn(SKIP_MSG))
-                .forEach(newDMaaPConf -> compareAndOverwriteDMaaPConfig(oldDMaaPConf, newDMaaPConf)));
-    }
-
-
-    private void updateConfigurationProperties(JSONObject newConf) {
-        configFilesFacade.readCollectorProperties()
-            .onFailure(logSkip())
-            .onSuccess(oldProps -> compareAndOverwritePropertiesConfig(newConf, oldProps));
-    }
-
-    private void compareAndOverwritePropertiesConfig(JSONObject newConf, Map<String, String> oldProps) {
-        Map<String, String> newProperties = getProperties(newConf);
-        Map<String, String> result = oldProps.filterKeys((s) ->  newProperties.keySet().contains(s));
-        if (!result.equals(newProperties)) {
-            configFilesFacade.writeProperties(newProperties)
-                .onSuccess(__ -> {
-                    toRestart= true;
-                    log.info("New properties configuration written to file");
-                 })
-                .onFailure(logSkip());
-        } else {
-            log.info("Collector properties from CBS are the same as currently used ones. " + SKIP_MSG);
-        }
-    }
-
-    private void compareAndOverwriteDMaaPConfig(JSONObject oldDMaaPConf, JSONObject newDMaaPConf) {
-        if (!oldDMaaPConf.toString().equals(newDMaaPConf.toString())) {
-            parseToDomainMapping(newDMaaPConf)
-                .onFailure(exc -> log.error(SKIP_MSG, exc))
-                .onSuccess(eventPublisherReconfigurer)
-                .onSuccess(parsedConfig ->
-                    configFilesFacade.writeDMaaPConfiguration(newDMaaPConf)
-                        .onFailure(logSkip())
-                        .onSuccess(__ -> {
-                            toRestart= true;
-                            log.info("New dMaaP configuration written to file");
-                        }));
-        } else {
-            log.info("DMaaP config from CBS is the same as currently used one. " + SKIP_MSG);
-        }
-    }
-
-    private Consumer<Throwable> logSkip() {
-        return __ -> log.error(SKIP_MSG);
-    }
-}
diff --git a/src/main/java/org/onap/dcae/controller/ConfigSource.java b/src/main/java/org/onap/dcae/controller/ConfigSource.java
deleted file mode 100644
index a9e439e4..00000000
--- a/src/main/java/org/onap/dcae/controller/ConfigSource.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.controller;
-
-import static io.vavr.API.Try;
-import static org.onap.dcae.common.publishing.VavrUtils.enhanceError;
-import static org.onap.dcae.common.publishing.VavrUtils.f;
-import static org.onap.dcae.controller.Conversions.toJson;
-import static org.onap.dcae.controller.Conversions.toJsonArray;
-
-import com.mashape.unirest.http.HttpResponse;
-import com.mashape.unirest.http.Unirest;
-import io.vavr.control.Try;
-import org.json.JSONArray;
-import org.json.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-final class ConfigSource {
-
-    private static final Logger log = LoggerFactory.getLogger(ConfigSource.class);
-
-    static Try<JSONObject> getAppConfig(EnvProps envProps) {
-        log.info("Fetching app configuration from CBS");
-        return callConsulForCBSConfiguration(envProps)
-            .peek(strBody -> log.info(f("Received following CBS configuration from Consul '%s'", strBody)))
-            .flatMap(Conversions::toJsonArray)
-            .flatMap(ConfigSource::withdrawCatalog)
-            .flatMap(json -> constructFullCBSUrl(envProps, json))
-            .flatMap(cbsUrl -> callCBSForAppConfig(envProps, cbsUrl))
-            .flatMap(Conversions::toJson)
-            .peek(jsonNode -> log.info(f("Received app configuration: '%s'", jsonNode)))
-            .onFailure(exc -> log.error("Could not fetch application config", exc));
-    }
-
-    private static Try<String> callConsulForCBSConfiguration(EnvProps envProps) {
-        return executeGet(envProps.consulProtocol + "://" + envProps.consulHost + ":" +
-            envProps.consulPort + "/v1/catalog/service/" + envProps.cbsName)
-            .mapFailure(enhanceError("Unable to retrieve CBS configuration from Consul"));
-    }
-
-    private static Try<String> constructFullCBSUrl(EnvProps envProps, JSONObject json) {
-        return Try(() -> envProps.cbsProtocol + "://" + json.get("ServiceAddress").toString() + ":" +
-            json.get("ServicePort").toString())
-            .mapFailure(enhanceError("ServiceAddress / ServicePort missing from CBS conf: '%s'", json));
-    }
-
-    private static Try<JSONObject> withdrawCatalog(JSONArray json) {
-        return Try(() -> new JSONObject(json.get(0).toString()))
-            .mapFailure(enhanceError("CBS response '%s' is in invalid format,"
-                + " most probably is it not a list of configuration objects", json));
-    }
-
-    private static Try<String> callCBSForAppConfig(EnvProps envProps, String cbsUrl) {
-        log.info("Calling CBS for application config");
-        return executeGet(cbsUrl + "/service_component/" + envProps.appName)
-            .mapFailure(enhanceError("Unable to fetch configuration from CBS"));
-    }
-
-
-    private static Try<String> executeGet(String url) {
-        log.info(f("Calling HTTP GET on url: '%s'", url));
-        return Try(() -> Unirest.get(url).asString())
-            .mapFailure(enhanceError("Http call (GET '%s') failed.", url))
-            .filter(
-                res -> res.getStatus() == 200,
-                res -> new RuntimeException(f("HTTP call (GET '%s') failed with status %s and body '%s'",
-                    url, res.getStatus(), res.getBody())))
-            .map(HttpResponse::getBody)
-            .peek(body -> log.info(f("HTTP GET on '%s' returned body '%s'", url, body)));
-    }
-
-}
diff --git a/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java b/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java
deleted file mode 100644
index 319caa65..00000000
--- a/src/main/java/org/onap/dcae/controller/EnvPropertiesReader.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.controller;
-
-import io.vavr.collection.Map;
-import io.vavr.control.Option;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static io.vavr.API.List;
-import static io.vavr.API.Try;
-import static org.onap.dcae.common.publishing.VavrUtils.f;
-
-final class EnvPropertiesReader {
-
-    private final static Logger log = LoggerFactory.getLogger(EnvPropertiesReader.class);
-
-    static Option<EnvProps> readEnvProps(Map<String, String> environmentVariables) {
-        log.info("Loading necessary environment variables for dynamic configuration update");
-        int consulPort = getConsulPort(environmentVariables);
-        String consulProtocol = getConsulProtocol(environmentVariables);
-        String cbsProtocol = getCbsProtocol(environmentVariables);
-        Option<String> consulHost = getConsulHost(environmentVariables);
-        Option<String> cbsServiceName = getCBSName(environmentVariables);
-        Option<String> vesCollectorAppName = getAppName(environmentVariables);
-        return Option.sequence(List(consulHost, cbsServiceName, vesCollectorAppName))
-            .map(e -> new EnvProps(consulProtocol, e.get(0), consulPort, cbsProtocol, e.get(1), e.get(2)))
-            .onEmpty(() -> log.warn("Some required environment variables are missing"))
-            .peek(props -> log.info(f("Discovered following environment variables: '%s'", props)));
-    }
-
-    private static Option<String> getAppName(Map<String, String> environmentVariables) {
-        return environmentVariables.get("HOSTNAME")
-            .orElse(environmentVariables.get("SERVICE_NAME"))
-            .onEmpty(() -> log.warn("App service name (as registered in CBS) (env var: 'HOSTNAME' / 'SERVICE_NAME') "
-                + "is missing error environment variables."));
-    }
-
-    private static Option<String> getCBSName(Map<String, String> environmentVariables) {
-        return environmentVariables.get("CONFIG_BINDING_SERVICE")
-            .onEmpty(() -> log.warn("Name of CBS Service (as registered in Consul) (env var: 'CONFIG_BINDING_SERVICE') "
-                + "is missing from environment variables."));
-    }
-
-    private static Integer getConsulPort(Map<String, String> environmentVariables) {
-        return environmentVariables.get("CONSUL_PORT")
-            .flatMap(str -> Try(() -> Integer.valueOf(str))
-                .onFailure(exc -> log.warn("Consul port is not an integer value", exc))
-                .toOption())
-            .onEmpty(() -> log.warn("Consul port (env var: 'CONSUL_PORT') is missing from environment variables. "
-                + "Using default value of 8500"))
-            .getOrElse(8500);
-    }
-
-    private static Option<String> getConsulHost(Map<String, String> environmentVariables) {
-        return environmentVariables.get("CONSUL_HOST")
-            .onEmpty(() -> log.warn("Consul host (env var: 'CONSUL_HOST') (without port) "
-                + "is missing from environment variables."));
-    }
-
-    private static String getConsulProtocol(Map<String, String> environmentVariables) {
-        return getProtocolFrom("CONSUL_PROTOCOL", environmentVariables);
-    }
-
-    private static String getCbsProtocol(Map<String, String> environmentVariables) {
-        return getProtocolFrom("CBS_PROTOCOL", environmentVariables);
-    }
-
-    private static String getProtocolFrom(String variableName, Map<String, String> environmentVariables) {
-        return environmentVariables.get(variableName)
-            .onEmpty(() -> log.warn("Consul protocol (env var: '" + variableName + "') is missing "
-                + "from environment variables."))
-            .getOrElse("http");
-    }
-
-}
diff --git a/src/main/java/org/onap/dcae/controller/EnvProps.java b/src/main/java/org/onap/dcae/controller/EnvProps.java
deleted file mode 100644
index 5f7d08d5..00000000
--- a/src/main/java/org/onap/dcae/controller/EnvProps.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.controller;
-
-import java.util.Objects;
-
-/**
- * @author Pawel Szalapski (pawel.szalapski@nokia.com)
- */
-final class EnvProps {
-
-    final String consulProtocol;
-    final String consulHost;
-    final int consulPort;
-    final String cbsName;
-    final String cbsProtocol;
-    final String appName;
-
-    EnvProps(String consulProtocol, String consulHost, int consulPort, String cbsProtocol, String cbsName, String appName) {
-        this.consulProtocol = consulProtocol;
-        this.consulHost = consulHost;
-        this.consulPort = consulPort;
-        this.cbsProtocol = cbsProtocol;
-        this.cbsName = cbsName;
-        this.appName = appName;
-    }
-
-    @Override
-    public String toString() {
-        return "EnvProps{" +
-            "consulProtocol='" + consulProtocol + '\'' +
-            ", consulHost='" + consulHost + '\'' +
-            ", consulPort=" + consulPort +
-            ", cbsProtocol='" + cbsProtocol + '\'' +
-            ", cbsName='" + cbsName + '\'' +
-            ", appName='" + appName + '\'' +
-            '}';
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        EnvProps envProps = (EnvProps) o;
-        return consulPort == envProps.consulPort &&
-            Objects.equals(consulProtocol, envProps.consulProtocol) &&
-            Objects.equals(consulHost, envProps.consulHost) &&
-            Objects.equals(cbsProtocol, envProps.cbsProtocol) &&
-            Objects.equals(cbsName, envProps.cbsName) &&
-            Objects.equals(appName, envProps.appName);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(consulProtocol, consulHost, consulPort, cbsProtocol, cbsName, appName);
-    }
-}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/TLSTestBase.java b/src/test/java/org/onap/dcae/TLSTestBase.java
index 1eb5728e..1026e759 100644
--- a/src/test/java/org/onap/dcae/TLSTestBase.java
+++ b/src/test/java/org/onap/dcae/TLSTestBase.java
@@ -2,9 +2,8 @@
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018-2020 Nokia. All rights reserved.
  * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2019 Nokia. 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.
@@ -136,13 +135,6 @@ public class TLSTestBase {
             return createHttpsRestTemplate().getForEntity(createHttpsURL("/"), String.class);
         }
 
-
-        public ResponseEntity<String> makeHttpsRequestWithBasicAuth(final String username, final String password) {
-            return addBasicAuth(createHttpsRestTemplate(), username, password)
-                    .getForEntity(createHttpsURL("/"), String.class);
-
-        }
-
         public ResponseEntity<String> makeHttpsRequestWithClientCert() {
             return createHttpsRestTemplateWithKeyStore().getForEntity(createHttpsURL("/"), String.class);
         }
diff --git a/src/test/java/org/onap/dcae/TestingUtilities.java b/src/test/java/org/onap/dcae/TestingUtilities.java
index 4c0d5382..8fb89154 100644
--- a/src/test/java/org/onap/dcae/TestingUtilities.java
+++ b/src/test/java/org/onap/dcae/TestingUtilities.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. All rights reserved.
  * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -20,23 +20,7 @@
  */
 package org.onap.dcae;
 
-import static java.nio.file.Files.readAllBytes;
-import static org.assertj.core.api.Assertions.assertThat;
-
 import io.vavr.control.Try;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.security.KeyStore;
-import java.security.KeyStoreException;
-import java.security.NoSuchAlgorithmException;
-import java.security.UnrecoverableKeyException;
-import java.security.cert.CertificateException;
-
 import org.apache.http.client.HttpClient;
 import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
 import org.apache.http.impl.client.HttpClients;
@@ -48,6 +32,15 @@ import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
 import org.springframework.web.client.RestTemplate;
 
 import javax.net.ssl.SSLContext;
+import java.io.File;
+import java.io.FileInputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.KeyStore;
+
+import static java.nio.file.Files.readAllBytes;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Pawel Szalapski (pawel.szalapski@nokia.com)
diff --git a/src/test/java/org/onap/dcae/WiremockBasedTest.java b/src/test/java/org/onap/dcae/WiremockBasedTest.java
deleted file mode 100644
index 3b4bdc39..00000000
--- a/src/test/java/org/onap/dcae/WiremockBasedTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
- * Copyright (C) 2018 AT&T Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.dcae;
-
-import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
-import static com.github.tomakehurst.wiremock.client.WireMock.get;
-import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
-import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
-import static io.vavr.API.Map;
-
-import com.github.tomakehurst.wiremock.junit.WireMockRule;
-import io.vavr.collection.Map;
-import org.junit.Rule;
-
-/**
- * @author Pawel Szalapski (pawel.szalapski@nokia.com)
- */
-public class WiremockBasedTest {
-
-    @Rule
-    public WireMockRule wireMockRule = new WireMockRule(
-        wireMockConfig().dynamicPort().dynamicHttpsPort().keystorePath(null));
-
-    protected void stubConsulToReturnLocalAddressOfCBS() {
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withBody(validLocalCBSConf())));
-    }
-
-    protected void stubCBSToReturnAppConfig(String sampleConfigForVES) {
-        stubFor(get(urlEqualTo("/service_component/VESCollector"))
-            .willReturn(aResponse().withBody(sampleConfigForVES)));
-    }
-
-    protected Map<String, String> wiremockBasedEnvProps() {
-        return Map(
-            "CONSUL_HOST", "localhost",
-            "CONSUL_PORT", "" + wireMockRule.port(),
-            "HOSTNAME", "VESCollector",
-            "CONFIG_BINDING_SERVICE", "CBSName"
-        );
-    }
-
-    protected String validLocalCBSConf() {
-        return ""
-            + "[{ "
-            + "\"ServiceAddress\": \"localhost\","
-            + "\"ServicePort\":" + wireMockRule.port()
-            + "}]";
-    }
-}
diff --git a/src/test/java/org/onap/dcae/common/EventSenderTest.java b/src/test/java/org/onap/dcae/common/EventSenderTest.java
index e20fa993..454cfb52 100644
--- a/src/test/java/org/onap/dcae/common/EventSenderTest.java
+++ b/src/test/java/org/onap/dcae/common/EventSenderTest.java
@@ -3,7 +3,7 @@
  * PROJECT
  * ================================================================================
  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2020 Nokia. All rights reserved.s
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -28,7 +28,7 @@ import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.onap.dcae.common.model.StndDefinedNamespaceParameterNotDefinedException;
 import org.onap.dcae.common.model.VesEvent;
-import org.onap.dcae.common.publishing.EventPublisher;
+import org.onap.dcae.common.publishing.DMaaPEventPublisher;
 
 import java.io.IOException;
 import java.util.List;
@@ -43,7 +43,7 @@ import static org.mockito.Mockito.verify;
 public class EventSenderTest {
 
   @Mock
-  private EventPublisher eventPublisher;
+  private DMaaPEventPublisher eventPublisher;
 
 
   @Test
diff --git a/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java b/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java
index 45cdf282..99505bfd 100644
--- a/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java
+++ b/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. All rights reserved.
  * Copyright (C) 2020 AT&T. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -47,7 +47,7 @@ public class DMaaPEventPublisherTest {
         cambriaPublisher = mock(CambriaBatchingPublisher.class);
         DMaaPPublishersCache = mock(DMaaPPublishersCache.class);
         when(DMaaPPublishersCache.getPublisher(anyString())).thenReturn(Option(cambriaPublisher));
-        eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache, mock(Logger.class));
+        eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache);
     }
 
     @Test
diff --git a/src/test/java/org/onap/dcae/controller/ConfigFilesFacadeTest.java b/src/test/java/org/onap/dcae/configuration/ConfigFilesFacadeTest.java
index 474a77c5..8849369a 100644
--- a/src/test/java/org/onap/dcae/controller/ConfigFilesFacadeTest.java
+++ b/src/test/java/org/onap/dcae/configuration/ConfigFilesFacadeTest.java
@@ -3,7 +3,7 @@
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -18,7 +18,7 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.controller;
+package org.onap.dcae.configuration;
 
 import static io.vavr.API.Map;
 import static io.vavr.API.Some;
@@ -134,6 +134,4 @@ public class ConfigFilesFacadeTest {
         assertThat(result.isFailure()).isTrue();
         assertFailureHasInfo(result, NON_EXISTENT.toString());
     }
-
-}
-
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/EnvPropsTest.java b/src/test/java/org/onap/dcae/configuration/ConfigLoaderFactoryTest.java
index ccc083e2..e9421909 100644
--- a/src/test/java/org/onap/dcae/controller/EnvPropsTest.java
+++ b/src/test/java/org/onap/dcae/configuration/ConfigLoaderFactoryTest.java
@@ -1,4 +1,3 @@
-package org.onap.dcae.controller;
 /*-
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
@@ -18,22 +17,27 @@ package org.onap.dcae.controller;
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-import org.jetbrains.annotations.NotNull;
-import org.junit.Test;
+package org.onap.dcae.configuration;
+
 import static org.assertj.core.api.Assertions.assertThat;
-public class EnvPropsTest {
+
+import java.nio.file.Path;
+import org.junit.Test;
+
+public class ConfigLoaderFactoryTest {
+
     @Test
-    public void shouldBeEquals() {
+    public void createsCbsConfigLoaderSuccessfully() {
         // given
-        EnvProps envPropsOriginal = givenEnvProps();
-        EnvProps envPropsCopy = givenEnvProps();
-        // when/then
-        assertThat(envPropsOriginal).isEqualTo(envPropsCopy);
-        assertThat(envPropsOriginal.hashCode()).isEqualTo(envPropsCopy.hashCode());
-    }
-    @NotNull
-    private EnvProps givenEnvProps() {
-        return new EnvProps("https", "localhost", 443,
-                "https", "cbsName", "appName");
+        Path testPropertiesPath = Path.of("src/test/resources/testcollector.properties");
+        Path testDmaapConfigPath = Path.of("src/test/resources/testParseDMaaPCredentialsGen2.json");
+
+        // when
+        ConfigLoader configLoader = ConfigLoaderFactory.create(
+            testPropertiesPath,
+            testDmaapConfigPath);
+
+        // then
+        assertThat(configLoader).isNotNull();
     }
-}
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java b/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java
new file mode 100644
index 00000000..46d864a0
--- /dev/null
+++ b/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java
@@ -0,0 +1,294 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.onap.dcae.TestingUtilities.readJSONFromFile;
+import static org.onap.dcae.common.publishing.VavrUtils.f;
+
+import io.vavr.collection.HashMap;
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.nio.file.Paths;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dcae.configuration.cbs.CbsConfigResolver;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class ConfigLoaderTest {
+
+    private static final String COLLECTOR_PORT = "collector.port";
+    private static final String PORT_8080 = "8080";
+    private static final String PORT_8081 = "8081";
+    private static final String COLLECTOR_KEYSTORE_FILE_LOCATION = "collector.keystore.file.location";
+    private static final String SOME_PATH = "some/path";
+    private static final String COLLECTOR_SCHEMA_FILE = "collector.schema.file";
+    private static final String SOME_SCHEMA = "some schema";
+
+    @Mock
+    private CbsConfigResolver cbsConfigResolverMock;
+
+    @Mock
+    private ConfigFilesFacade configFilesFacadeMock;
+
+    @InjectMocks
+    private ConfigLoader configLoader;
+
+    @Mock
+    private Runnable applicationRestarter;
+
+
+    @Before
+    public void setup() {
+        when(configFilesFacadeMock.readCollectorProperties()).thenReturn(Try.of(HashMap::empty));
+        when(configFilesFacadeMock.readDMaaPConfiguration()).thenReturn(Try.of(JSONObject::new));
+    }
+
+    @Test
+    public void shouldCallConfigSourceForData() {
+        // given
+        HashMap<String, String> properties = HashMap.of(COLLECTOR_PORT, PORT_8080);
+        mockVesInitialProperties(properties);
+        mockVesConfigInCbs(properties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(cbsConfigResolverMock).getAppConfig();
+    }
+
+    @Test
+    public void shouldNotUpdatePropertiesWhenSameKeySetAndSameValues() {
+        // given
+        HashMap<String, String> properties = HashMap.of(COLLECTOR_PORT, PORT_8080);
+        mockVesInitialProperties(properties);
+        mockVesConfigInCbs(properties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, never()).writeProperties(any());
+        verify(applicationRestarter, never()).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenSameKeySetButDifferentValues() {
+        // given
+        HashMap<String, String> initialProperties = HashMap.of(COLLECTOR_PORT, PORT_8080);
+        HashMap<String, String> cbsProperties = HashMap.of(COLLECTOR_PORT, PORT_8081);
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenVesKeysAreSubsetOfCbsKeysAndSubsetHasSameValues() {
+        // given
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenVesKeysAreSubsetOfCbsKeysAndSubsetHasDifferentValues() {
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8081,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldNotUpdatePropertiesWhenCbsKeysAreSubsetOfVesKeysAndSubsetHasSameValues() {
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080);
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, never()).writeProperties(any());
+        verify(applicationRestarter, never()).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenCbsKeysAreSubsetOfVesKeysAndSubsetHasDifferentValues() {
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8081);
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenVesAndCbsKeySetsIntersectAndIntersectingKeysHaveSameValues() {
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_SCHEMA_FILE, SOME_SCHEMA
+        );
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldUpdatePropertiesWhenVesAndCbsKeySetsIntersectAndIntersectingKeysHaveDifferentValues() {
+        HashMap<String, String> initialProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8080,
+            COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
+        HashMap<String, String> cbsProperties = HashMap.of(
+            COLLECTOR_PORT, PORT_8081,
+            COLLECTOR_SCHEMA_FILE, SOME_SCHEMA
+        );
+        mockVesInitialProperties(initialProperties);
+        mockVesConfigInCbs(cbsProperties);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldUpdateDmaapConfigWhenConfigurationChanged() {
+        // given
+        JSONObject emptyDmaapConfig = new JSONObject();
+        JSONObject dmaapConfig = loadSampleDmaapConfig();
+        mockVesInitialDmaapConfig(emptyDmaapConfig);
+        mockVesDmaapConfigInCbs(dmaapConfig);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock).writeDMaaPConfiguration(argThat(dmaapConfig::similar));
+        verify(applicationRestarter, times(1)).run();
+    }
+
+    @Test
+    public void shouldNotUpdateDmaapConfigWhenConfigurationNotChanged() {
+        // given
+        JSONObject dmaapConf = loadSampleDmaapConfig();
+        mockVesInitialDmaapConfig(dmaapConf);
+        mockVesDmaapConfigInCbs(dmaapConf);
+
+        // when
+        configLoader.updateConfig();
+
+        // then
+        verify(configFilesFacadeMock, never()).writeDMaaPConfiguration(any());
+        verify(applicationRestarter, never()).run();
+    }
+
+    private void mockVesInitialDmaapConfig(JSONObject dmaapConf) {
+        when(configFilesFacadeMock.readDMaaPConfiguration()).thenReturn(Try.of(() -> dmaapConf));
+    }
+
+    private void mockVesDmaapConfigInCbs(JSONObject dmaapConf) {
+        JSONObject jsonObject = new JSONObject(f("{\"streams_publishes\": %s}}", dmaapConf));
+        when(cbsConfigResolverMock.getAppConfig()).thenReturn(Option.of(jsonObject));
+    }
+
+    private void mockVesConfigInCbs(HashMap<String, String> properties) {
+        when(cbsConfigResolverMock.getAppConfig()).thenReturn(Option.of(prepareConfigurationJson(properties)));
+    }
+
+    private void mockVesInitialProperties(HashMap<String, String> properties) {
+        when(configFilesFacadeMock.readCollectorProperties()).thenReturn(Try.of(() -> properties));
+    }
+
+
+    private JSONObject loadSampleDmaapConfig() {
+        return readJSONFromFile(Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json"));
+    }
+
+    private JSONObject prepareConfigurationJson(Map<String, String> properties) {
+        String template = "{%s, \"streams_publishes\": {}}";
+        String customProperties = properties
+            .map(property -> "\"" + property._1 + "\": \"" + property._2 + "\"")
+            .mkString(", ");
+        String jsonBody = f(template, customProperties);
+        return new JSONObject(jsonBody);
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/ConfigParsingTest.java b/src/test/java/org/onap/dcae/configuration/ConfigParsingTest.java
index 5d8d4efa..248d3c5a 100644
--- a/src/test/java/org/onap/dcae/controller/ConfigParsingTest.java
+++ b/src/test/java/org/onap/dcae/configuration/ConfigParsingTest.java
@@ -3,7 +3,7 @@
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
  * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2018,2020 Nokia. 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.
@@ -19,20 +19,20 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.dcae.controller;
+package org.onap.dcae.configuration;
 
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.junit.Test;
+
+import java.nio.file.Paths;
 
 import static io.vavr.API.Map;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.onap.dcae.TestingUtilities.assertJSONObjectsEqual;
 import static org.onap.dcae.TestingUtilities.readJSONFromFile;
 
-import io.vavr.collection.Map;
-import io.vavr.control.Option;
-import java.nio.file.Paths;
-import org.json.JSONObject;
-import org.junit.Test;
-
 public class ConfigParsingTest {
 
     @Test
diff --git a/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolverTest.java b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolverTest.java
new file mode 100644
index 00000000..57824595
--- /dev/null
+++ b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolverTest.java
@@ -0,0 +1,47 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration.cbs;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.Test;
+import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+
+public class CbsClientConfigurationResolverTest {
+
+    private static final String DEFAULT_PROTOCOL = "http";
+    private static final String DEFAULT_HOSTNAME = "config-binding-service";
+    private static final int DEFAULT_PORT = 10000;
+    private static final String DEFAULT_APP_NAME = "dcae-ves-collector";
+
+    @Test
+    @DisabledIfEnvironmentVariable(named = "CONFIG_BINDING_SERVICE", matches = ".+")
+    public void shouldLoadDefaultConfigWhenEnvNotPresent() {
+        // when
+        CbsClientConfiguration configuration = new CbsClientConfigurationResolver().resolveCbsClientConfiguration();
+
+        // then
+        assertThat(configuration.protocol()).isEqualTo(DEFAULT_PROTOCOL);
+        assertThat(configuration.hostname()).isEqualTo(DEFAULT_HOSTNAME);
+        assertThat(configuration.port()).isEqualTo(DEFAULT_PORT);
+        assertThat(configuration.appName()).isEqualTo(DEFAULT_APP_NAME);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/onap/dcae/common/publishing/EventPublisher.java b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientFactoryTest.java
index 42e721a8..4b17f68d 100644
--- a/src/main/java/org/onap/dcae/common/publishing/EventPublisher.java
+++ b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientFactoryTest.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * org.onap.dcaegen2.collectors.ves
  * ================================================================================
- * Copyright (C) 2018 Nokia. All rights reserved.
+ * Copyright (C) 2020 Nokia. 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.
@@ -17,22 +17,20 @@
  * limitations under the License.
  * ============LICENSE_END=========================================================
  */
-package org.onap.dcae.common.publishing;
+package org.onap.dcae.configuration.cbs;
 
-import io.vavr.collection.Map;
-import org.json.JSONObject;
-import org.slf4j.Logger;
+import static org.assertj.core.api.Assertions.assertThat;
 
-/**
- * @author Pawel Szalapski (pawel.szalapski@nokia.com)
- */
-public interface EventPublisher {
+import org.junit.Test;
 
-    static EventPublisher createPublisher(Logger outputLogger, Map<String, PublisherConfig> dMaaPConfig) {
-        return new DMaaPEventPublisher(new DMaaPPublishersCache(dMaaPConfig), outputLogger);
-    }
+public class CbsClientFactoryTest {
 
-    void sendEvent(JSONObject event, String domain);
+    @Test
+    public void createsClientSuccessfully() {
+        // when
+        CbsConfigResolver cbsConfigResolver = new CbsConfigResolverFactory().create();
 
-    void reconfigure(Map<String, PublisherConfig> dMaaPConfig);
-}
+        // then
+        assertThat(cbsConfigResolver).isNotNull();
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/configuration/cbs/CbsConfigResolverTest.java b/src/test/java/org/onap/dcae/configuration/cbs/CbsConfigResolverTest.java
new file mode 100644
index 00000000..b413a64d
--- /dev/null
+++ b/src/test/java/org/onap/dcae/configuration/cbs/CbsConfigResolverTest.java
@@ -0,0 +1,74 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.configuration.cbs;
+
+import com.github.tomakehurst.wiremock.junit.WireMockRule;
+import org.json.JSONObject;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableCbsClientConfiguration;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class CbsConfigResolverTest {
+
+    private static final String VES_CONFIG = "{\"collector.port\": 8081}";
+    private static final String HOSTNAME = "localhost";
+    private static final String PROTOCOL = "http";
+    private static final String APP_NAME = "VESCollector";
+
+    @Rule
+    public final WireMockRule wireMockRule = new WireMockRule(
+            wireMockConfig().dynamicPort().dynamicPort());
+
+    @Test
+    public void shouldFetchConfigurationFromCBS() {
+        // given
+        final int PORT = wireMockRule.port();
+        stubCBSToReturnAppConfig();
+
+        // when
+        CbsClientConfiguration cbsClientConfiguration = ImmutableCbsClientConfiguration.builder()
+                .protocol(PROTOCOL)
+                .hostname(HOSTNAME)
+                .port(PORT)
+                .appName(APP_NAME)
+                .build();
+        JSONObject appConfig = new CbsConfigResolver(cbsClientConfiguration).getAppConfig().get();
+
+        // then
+        assertThat(appConfig).isNotNull();
+        assertThat(appConfig.toString()).isEqualTo(new JSONObject(VES_CONFIG).toString());
+    }
+
+    private void stubCBSToReturnAppConfig() {
+        stubFor(get(urlEqualTo("/service_component/VESCollector"))
+                .willReturn(aResponse().withBody(CbsConfigResolverTest.VES_CONFIG)));
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java b/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java
deleted file mode 100644
index bfec3d97..00000000
--- a/src/test/java/org/onap/dcae/controller/ConfigCBSSourceTest.java
+++ /dev/null
@@ -1,153 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.dcae.controller;
-
-import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
-import static com.github.tomakehurst.wiremock.client.WireMock.get;
-import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
-import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
-import static org.assertj.core.api.Java6Assertions.assertThat;
-import static org.onap.dcae.TestingUtilities.assertFailureHasInfo;
-import static org.onap.dcae.controller.ConfigSource.getAppConfig;
-
-import io.vavr.control.Try;
-import org.json.JSONObject;
-import org.junit.Test;
-import org.onap.dcae.WiremockBasedTest;
-
-
-public class ConfigCBSSourceTest extends WiremockBasedTest {
-
-    @Test
-    public void shouldReturnValidAppConfiguration() {
-        // given
-        String sampleConfigForVES = "{\"collector.port\": 8080}";
-
-        stubConsulToReturnLocalAddressOfCBS();
-        stubCBSToReturnAppConfig(sampleConfigForVES);
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertThat(actual.get().toString()).isEqualTo(new JSONObject(sampleConfigForVES).toString());
-    }
-
-    @Test
-    public void shouldReturnFailureOnFailureToCommunicateWithConsul() {
-        // given
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withStatus(400)));
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "HTTP", "Consul", "400",
-            "http://localhost:" + wireMockRule.port() + "/v1/catalog/service/CBSName");
-    }
-
-    @Test
-    public void shouldReturnFailureOnBadJsonFromConsul() {
-        // given
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withStatus(200).withBody("[{")));
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "JSON", "array", "[{");
-    }
-
-    @Test
-    public void shouldReturnFailureOnInvalidCatalogFormat() {
-        // given
-        String notAListCatalog = ""
-            + "{"
-            + "\"ServiceAddress\":\"localhost\","
-            + "\"ServicePort\":" + wireMockRule.port()
-            + "}";
-
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withStatus(200).withBody(notAListCatalog)));
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "JSON", "array", notAListCatalog);
-    }
-
-
-    @Test
-    public void shouldReturnFailureIfConfigIsMissingRequiredProperties() {
-        // given
-        String actualConf = "{\"ServicePort\":" + wireMockRule.port() + "}";
-        String asCatalog = "[" + actualConf + "]";
-
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withStatus(200).withBody(asCatalog)));
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "ServiceAddress", "ServicePort", "missing", actualConf);
-    }
-
-
-    @Test
-    public void shouldReturnFailureOnFailureToCommunicateWithCBS() {
-        // given
-        stubFor(get(urlEqualTo("/v1/catalog/service/CBSName"))
-            .willReturn(aResponse().withStatus(200).withBody(validLocalCBSConf())));
-        stubFor(get(urlEqualTo("/service_component/VESCollector"))
-            .willReturn(aResponse().withStatus(400)));
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "HTTP", "CBS", "400",
-            "http://localhost:" + wireMockRule.port() + "/service_component/VESCollector");
-    }
-
-    @Test
-    public void shouldReturnFailureIfAppIsInvalidJsonDocument() {
-        // given
-        String invalidAppConf = "[$";
-        stubConsulToReturnLocalAddressOfCBS();
-        stubCBSToReturnAppConfig(invalidAppConf);
-
-        // when
-        Try<JSONObject> actual = tryToGetConfig();
-
-        // then
-        assertFailureHasInfo(actual, "JSON", "document", invalidAppConf);
-    }
-
-    private Try<JSONObject> tryToGetConfig() {
-        return getAppConfig(new EnvProps("http", "localhost", wireMockRule.port(), "http", "CBSName", "VESCollector"));
-    }
-}
-
diff --git a/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java b/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java
deleted file mode 100644
index 90c8a9c8..00000000
--- a/src/test/java/org/onap/dcae/controller/ConfigLoaderIntegrationE2ETest.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcae.controller;
-
-import static io.vavr.API.Map;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyZeroInteractions;
-import static org.mockito.Mockito.when;
-import static org.onap.dcae.TestingUtilities.createTemporaryFile;
-import static org.onap.dcae.TestingUtilities.readFile;
-import static org.onap.dcae.TestingUtilities.readJSONFromFile;
-import static org.onap.dcae.common.publishing.VavrUtils.f;
-
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import org.json.JSONObject;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-import org.onap.dcae.ApplicationSettings;
-import org.onap.dcae.WiremockBasedTest;
-import org.onap.dcae.common.publishing.DMaaPConfigurationParser;
-import org.onap.dcae.common.publishing.EventPublisher;
-
-@RunWith(MockitoJUnitRunner.Silent.class)
-public class ConfigLoaderIntegrationE2ETest extends WiremockBasedTest {
-
-    @Test
-    public void testSuccessfulE2EFlow() {
-        // given
-        Path dMaaPConfigFile = createTemporaryFile("{}");
-        Path collectorPropertiesFile = createTemporaryFile("");
-        Path dMaaPConfigSource = Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json");
-        JSONObject dMaaPConf = readJSONFromFile(dMaaPConfigSource);
-        stubConsulToReturnLocalAddressOfCBS();
-        stubCBSToReturnAppConfig(f("{\"collector.port\": 8080, \"streams_publishes\": %s}}", dMaaPConf));
-
-        EventPublisher eventPublisherMock = mock(EventPublisher.class);
-        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(dMaaPConfigFile, collectorPropertiesFile);
-        ConfigLoader configLoader = new ConfigLoader(eventPublisherMock::reconfigure, configFilesFacade, ConfigSource::getAppConfig, () -> wiremockBasedEnvProps());
-        configLoader.updateConfig();
-        // then
-        assertThat(readJSONFromFile(dMaaPConfigSource).toString()).isEqualTo(dMaaPConf.toString());
-        assertThat(readFile(collectorPropertiesFile).trim()).isEqualTo("collector.port = 8080");
-        verify(eventPublisherMock, times(1)).reconfigure(
-            DMaaPConfigurationParser.parseToDomainMapping(dMaaPConf).get()
-        );
-    }
-
-    @Test
-    public void shouldNotReconfigureNotOverwriteIfConfigurationHasNotChanged() {
-        // given
-        Path dMaaPConfigFile = createTemporaryFile("{}");
-        Path collectorPropertiesFile = createTemporaryFile("");
-        JSONObject dMaaPConf = readJSONFromFile(Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json"));
-        stubConsulToReturnLocalAddressOfCBS();
-        stubCBSToReturnAppConfig(f("{\"collector.port\": 8080, \"streams_publishes\": %s}}", dMaaPConf));
-        EventPublisher eventPublisherMock = mock(EventPublisher.class);
-        ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(dMaaPConfigFile, collectorPropertiesFile);
-        configFilesFacade.writeProperties(Map("collector.port", "8080"));
-        configFilesFacade.writeDMaaPConfiguration(dMaaPConf);
-
-        // when
-        ConfigLoader configLoader = new ConfigLoader(eventPublisherMock::reconfigure, configFilesFacade, ConfigSource::getAppConfig, () -> wiremockBasedEnvProps());
-        configLoader.updateConfig();
-
-        // then
-        verifyZeroInteractions(eventPublisherMock);
-    }
-
-}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java b/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java
deleted file mode 100644
index 6c358024..00000000
--- a/src/test/java/org/onap/dcae/controller/EnvPropertiesReaderTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * org.onap.dcaegen2.collectors.ves
- * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
- * Copyright (C) 2018 Nokia. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.dcae.controller;
-
-import static io.vavr.API.Map;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.onap.dcae.controller.EnvPropertiesReader.readEnvProps;
-
-import io.vavr.collection.Map;
-import org.junit.Test;
-
-
-public class EnvPropertiesReaderTest {
-
-    @Test
-    public void shouldReturnEmptyOnMissingConsulHost() {
-        Map<String, String> envs = Map(
-            "CONFIG_BINDING_SERVICE", "doesNotMatter",
-            "HOSTNAME", "doesNotMatter");
-        assertTrue(readEnvProps(envs).isEmpty());
-    }
-
-    @Test
-    public void shouldReturnEmptyOnMissingCBSName() {
-        Map<String, String> envs = Map(
-            "CONSUL_HOST", "doesNotMatter",
-            "HOSTNAME", "doesNotMatter");
-        assertTrue(readEnvProps(envs).isEmpty());
-    }
-
-    @Test
-    public void shouldReturnEmptyOnMissingVESAppName() {
-        Map<String, String> envs = Map(
-            "CONSUL_HOST", "doesNotMatter",
-            "CONFIG_BINDING_SERVICE", "doesNotMatter");
-        assertTrue(readEnvProps(envs).isEmpty());
-    }
-
-    @Test
-    public void shouldReturnSomeOfAllProperties() {
-        Map<String, String> envs = Map(
-            "CONSUL_HOST", "doesNotMatter",
-            "HOSTNAME", "doesNotMatter",
-            "CONFIG_BINDING_SERVICE", "doesNotMatter");
-        assertFalse(readEnvProps(envs).isEmpty());
-    }
-
-}
-
diff --git a/src/test/java/org/onap/dcae/restapi/VesRestControllerTest.java b/src/test/java/org/onap/dcae/restapi/VesRestControllerTest.java
index 765f2b43..0a03c1a4 100644
--- a/src/test/java/org/onap/dcae/restapi/VesRestControllerTest.java
+++ b/src/test/java/org/onap/dcae/restapi/VesRestControllerTest.java
@@ -41,7 +41,7 @@ import org.onap.dcae.common.EventTransformation;
 import org.onap.dcae.common.HeaderUtils;
 import org.onap.dcae.common.JsonDataLoader;
 import org.onap.dcae.common.validator.StndDefinedDataValidator;
-import org.onap.dcae.common.publishing.EventPublisher;
+import org.onap.dcae.common.publishing.DMaaPEventPublisher;
 import org.slf4j.Logger;
 import org.springframework.http.ResponseEntity;
 import org.springframework.mock.web.MockHttpServletRequest;
@@ -84,7 +84,7 @@ public class VesRestControllerTest {
     private HeaderUtils headerUtils;
 
     @Mock
-    private EventPublisher eventPublisher;
+    private DMaaPEventPublisher eventPublisher;
 
     @Mock
     private StndDefinedDataValidator stndDefinedDataValidator;
@@ -320,7 +320,7 @@ public class VesRestControllerTest {
         );
     }
 
-    private void verifyThatTransformedEventWasSend(EventPublisher eventPublisher, String eventBeforeTransformation) {
+    private void verifyThatTransformedEventWasSend(DMaaPEventPublisher eventPublisher, String eventBeforeTransformation) {
         // event before transformation
         assertThat(eventBeforeTransformation).contains("\"version\": \"4.0.1\"");
         assertThat(eventBeforeTransformation).contains("\"faultFieldsVersion\": \"4.0\"");
diff --git a/version.properties b/version.properties
index c3c1ab1c..53760fdc 100644
--- a/version.properties
+++ b/version.properties
@@ -1,6 +1,6 @@
 major=1
 minor=7
-patch=3
+patch=4
 base_version=${major}.${minor}.${patch}
 release_version=${base_version}
 snapshot_version=${base_version}-SNAPSHOT