Fetch configuration from CBS

- Fix memory leak.
- Add reactive configuration fetching from Consul. Now configuration is updated when any change in VES configuration has been done in Consul.

Change-Id: I9cd42e04844c9e99d4d03951185523b569dc9483
Issue-ID: DCAEGEN2-2495
Signed-off-by: Zebek Bogumil <bogumil.zebek@nokia.com>

21 files changed, 582 insertions, 343 deletions

diff --git a/Changelog.md b/Changelog.md
index 6fcb9bc4..5b08467b 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -37,4 +37,6 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## [1.7.7] - 29/09/2020
          - [DCAEGEN2-2462](https://jira.onap.org/browse/DCAEGEN2-2462) - Adapt schema-map.json and test files to updated 3GPP repos 
 ## [1.7.8] - 13/10/2020
-          - [DCAEGEN2-2478](https://jira.onap.org/browse/DCAEGEN2-2478) - Add logs from external-repo-manager lib
\ No newline at end of file
+          - [DCAEGEN2-2478](https://jira.onap.org/browse/DCAEGEN2-2478) - Add logs from external-repo-manager lib
+ # [1.7.9] - 01/11/2020
+         -  [DCAEGEN2-2495](https://jira.onap.org/browse/DCAEGEN2-2495) - Ves Collector is down because of java heap space
diff --git a/README.md b/README.md
index 7b93a989..a1733a23 100644
--- a/README.md
+++ b/README.md
@@ -38,7 +38,43 @@ Run the image using docker-compose.yml
 ```
 docker-compose up
 ```
-
+### Developer mode - run application from IDE
+
+To connect with "real" Consul instance you need to activate developer mode during starting VES application from IDE.
+Attention: Without developer mode (default mode) Ves started from IDE will not use Consul.
+
+1. Configure host mapping
+    
+    For Linux: In host file add mapping for config-binding-service
+    
+        vi /etc/hosts
+        SET_HERE_IP config-binding-service
+    
+2. At lab open port for config-binding-service
+
+    - Get basic information about config-binding-service
+    
+        ```
+        ubuntu@onap-7607-rke-node:~$ kubectl -n onap get services | grep config-binding-service
+        config-binding-service  ClusterIP  10.43.227.68    <none>  10000/TCP,10443/TCP  6d2h
+        ```
+    - Edit config-binding-service to change ClusterIP to NodePort to expose port
+ 
+        ```
+        kubectl -n onap edit service config-binding-service
+        ```
+      
+    - Get information about opened port for config-binding-service
+        
+        ```
+        ubuntu@onap-7607-rke-node:~$ kubectl -n onap get services | grep config-binding-service
+        config-binding-service  NodePort  10.43.227.68  <none>  10000:31029/TCP,10443:32719/TCP  6d2h
+        ```
+      
+3. Run application with properties 
+
+        -DdevMode=true -DcbsPort=31029
+        
 ### Generate auth credential
 
 Library to generate new cryptographic password is stored in dcaegen2/sdk -"security/crypt-password"
@@ -222,4 +258,4 @@ A client's certificate verification is disabled on the container by default; bel
         
     Example of authentication failure (without a client's certificate):
         curl -i -X POST -d @event.json --header "Content-Type: application/json" https://localhost:8443/eventListener/v7 -k
-        curl: (35) error:14094412:SSL routines:ssl3_read_bytes:sslv3 alert bad certificate
\ No newline at end of file
+        curl: (35) error:14094412:SSL routines:ssl3_read_bytes:sslv3 alert bad certificate
diff --git a/pom.xml b/pom.xml
index d4875c42..762980f2 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.8-SNAPSHOT</version>

+  <version>1.7.9-SNAPSHOT</version>

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

   <description>VESCollector</description>

   <properties>

@@ -206,7 +206,7 @@
             <!-- Maven is loosing file permissions during artifacts copy -->

             <run>chmod +x bin/*.sh</run>

             <run>

-              <![CDATA[apt-get update && apt-get install -y --no-install-recommends procps && apt-get install -y vim && apt-get install -y curl && apt-get clean all]]></run>

+              <![CDATA[apt-get update && apt-get install -y --no-install-recommends procps && apt-get install -y less && apt-get install -y vim && apt-get install -y curl && apt-get clean all]]></run>

           </runs>

           <exposes>

             <expose>8080</expose>

@@ -435,6 +435,12 @@
       <artifactId>spring-boot-starter-test</artifactId>

       <version>${spring-boot-starter-test.version}</version>

       <scope>test</scope>

+      <exclusions>

+        <exclusion>

+          <groupId>com.vaadin.external.google</groupId>

+          <artifactId>android-json</artifactId>

+        </exclusion>

+      </exclusions>

     </dependency>

     <dependency>

       <groupId>org.onap.dcaegen2.services.sdk.security.crypt</groupId>

diff --git a/src/main/java/org/onap/dcae/ApplicationConfigurationListener.java b/src/main/java/org/onap/dcae/ApplicationConfigurationListener.java
new file mode 100644
index 00000000..b86bc1ec
--- /dev/null
+++ b/src/main/java/org/onap/dcae/ApplicationConfigurationListener.java
@@ -0,0 +1,105 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * VES Collector
+ * ================================================================================
+ * 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;
+
+import org.onap.dcae.configuration.ConfigurationHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
+
+import java.time.Duration;
+
+/**
+ * ApplicationConfigurationListener is used to listen at notifications with configuration updates send from Consul.
+ */
+public class ApplicationConfigurationListener implements Runnable {
+
+    private static Logger log = LoggerFactory.getLogger(ApplicationConfigurationListener.class);
+
+    private Duration interval;
+    private boolean terminate = false;
+    private final ConfigurationHandler configurationHandler;
+
+    /**
+     * Constructor
+     * @param interval defines period of time when notification can come
+     * @param configurationHandler handles notifications send by Consul
+     */
+    public ApplicationConfigurationListener(Duration interval, ConfigurationHandler configurationHandler) {
+        this.interval = interval;
+        this.configurationHandler = configurationHandler;
+    }
+
+    /**
+     * Reload listener to start listening for Consul notifications with defined interval.
+     * @param interval defines period of time when notification can come
+     */
+    public synchronized void reload(Duration interval) {
+        this.interval = interval;
+        log.info("Handler configuration was changed. Need to reload configuration handler.");
+        sendReloadAction();
+    }
+
+    synchronized void sendReloadAction() {
+        this.notifyAll();
+    }
+
+    /**
+     * Start listening for Consul notification.
+     */
+    @Override
+    public void run() {
+        Disposable consulListener = null;
+        do {
+            try {
+                consulListener = listenForConfigurationUpdates();
+                synchronized (this) {
+                    log.info("Switch to configuration handler thread. Active waiting for configuration from Consul.");
+                    this.wait();
+                }
+            } catch (Exception e) {
+                log.error("Unexpected error occurred during handling data from Consul.", e);
+                terminate();
+            } finally {
+                stopListeningForConfigurationUpdates(consulListener);
+            }
+        } while (!this.terminate);
+    }
+
+    private Disposable listenForConfigurationUpdates() {
+        return this.configurationHandler.startListen(this.interval);
+    }
+
+    void terminate() {
+        this.terminate = true;
+    }
+
+    /**
+     * Release resources when there is a need to stop listener
+     * @param consulListener Handler to Consul listener
+     */
+    void stopListeningForConfigurationUpdates(Disposable consulListener) {
+        if (consulListener != null) {
+            consulListener.dispose();
+        }
+    }
+}
diff --git a/src/main/java/org/onap/dcae/VesApplication.java b/src/main/java/org/onap/dcae/VesApplication.java
index e5ee6e35..ec04157f 100644
--- a/src/main/java/org/onap/dcae/VesApplication.java
+++ b/src/main/java/org/onap/dcae/VesApplication.java
@@ -27,8 +27,10 @@ import org.onap.dcae.common.validator.StndDefinedValidatorResolver;
 import org.onap.dcae.common.publishing.DMaaPConfigurationParser;
 import org.onap.dcae.common.publishing.DMaaPEventPublisher;
 import org.onap.dcae.common.publishing.PublisherConfig;
-import org.onap.dcae.configuration.ConfigLoader;
-import org.onap.dcae.configuration.ConfigLoaderFactory;
+import org.onap.dcae.configuration.ConfigurationHandler;
+import org.onap.dcae.configuration.ConfigUpdater;
+import org.onap.dcae.configuration.ConfigUpdaterFactory;
+import org.onap.dcae.configuration.cbs.CbsClientConfigurationProvider;
 import org.onap.dcaegen2.services.sdk.services.external.schema.manager.service.StndDefinedValidator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,37 +44,38 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Lazy;
 
 import java.nio.file.Paths;
-import java.util.concurrent.ScheduledFuture;
+import java.time.Duration;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 
 @SpringBootApplication(exclude = {GsonAutoConfiguration.class, SecurityAutoConfiguration.class})
 public class VesApplication {
 
+    private static final int DEFAULT_CONFIGURATION_FETCH_PERIOD = 5;
+
     private static final Logger incomingRequestsLogger = LoggerFactory.getLogger("org.onap.dcae.common.input");
     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 ConfigUpdater configUpdater;
     private static DMaaPEventPublisher eventPublisher;
-    private static ScheduledFuture<?> scheduleFeatures;
+    private static ApplicationConfigurationListener applicationConfigurationListener;
 
     public static void main(String[] args) {
         SpringApplication app = new SpringApplication(VesApplication.class);
         applicationSettings = new ApplicationSettings(args, CLIUtils::processCmdLine);
-        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
         init();
+
+        applicationConfigurationListener = startListeningForApplicationConfigurationStoredInConsul();
+
         app.setAddCommandLineProperties(true);
         context = app.run();
-        configLoader.updateConfig();
     }
 
     public static void restartApplication() {
         Thread thread = new Thread(() -> {
             context.close();
             applicationSettings.reloadProperties();
-            scheduleFeatures.cancel(true);
+            applicationConfigurationListener.reload(Duration.ofMinutes(applicationSettings.configurationUpdateFrequency()));
             init();
             context = SpringApplication.run(VesApplication.class);
         });
@@ -81,26 +84,20 @@ public class VesApplication {
     }
 
     private static void init() {
-        createConfigLoader();
-        createSchedulePoolExecutor();
-        createExecutors();
-    }
-
-    private static void createExecutors() {
+        configUpdater = ConfigUpdaterFactory.create(
+                applicationSettings.configurationFileLocation(),
+                Paths.get(applicationSettings.dMaaPConfigurationFileLocation()));
         eventPublisher = new DMaaPEventPublisher(getDmaapConfig());
     }
 
-    private static void createSchedulePoolExecutor() {
-        scheduleFeatures = scheduledThreadPoolExecutor.scheduleAtFixedRate(configLoader::updateConfig,
-                applicationSettings.configurationUpdateFrequency(),
-                applicationSettings.configurationUpdateFrequency(),
-                TimeUnit.MINUTES);
-    }
+    private static ApplicationConfigurationListener startListeningForApplicationConfigurationStoredInConsul() {
+        ConfigurationHandler cbsHandler = new ConfigurationHandler(new CbsClientConfigurationProvider(), configUpdater);
+        ApplicationConfigurationListener applicationConfigProvider = new ApplicationConfigurationListener(Duration.ofMinutes(DEFAULT_CONFIGURATION_FETCH_PERIOD), cbsHandler);
 
-    private static void createConfigLoader() {
-        configLoader = ConfigLoaderFactory.create(
-                applicationSettings.configurationFileLocation(),
-                Paths.get(applicationSettings.dMaaPConfigurationFileLocation()));
+        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
+        scheduledThreadPoolExecutor.execute(applicationConfigProvider);
+
+        return applicationConfigProvider;
     }
 
     private static Map<String, PublisherConfig> getDmaapConfig() {
@@ -136,5 +133,4 @@ public class VesApplication {
     public StndDefinedValidator getStndDefinedValidator(StndDefinedValidatorResolver resolver) {
         return resolver.resolve();
     }
-
 }
diff --git a/src/main/java/org/onap/dcae/configuration/ConfigParsing.java b/src/main/java/org/onap/dcae/configuration/ConfigParsing.java
index 13deff73..d648332d 100644
--- a/src/main/java/org/onap/dcae/configuration/ConfigParsing.java
+++ b/src/main/java/org/onap/dcae/configuration/ConfigParsing.java
@@ -47,11 +47,11 @@ interface ConfigParsing {
     }
 
     static Map<String, String> getProperties(JSONObject configuration) {
-        log.info(f("Getting properties configuration from app configuration: '%s'", configuration));
+        log.debug(f("Getting properties configuration from app configuration: '%s'", configuration));
         Map<String, String> confEntries = toList(configuration.toMap().entrySet().iterator())
             .toMap(e -> Tuple(e.getKey(), String.valueOf(e.getValue())))
             .filterKeys(e -> !e.startsWith("streams_publishes"));
-        log.info(f("Found following app properties: '%s'", confEntries));
+        log.debug(f("Found following app properties: '%s'", confEntries));
         return confEntries;
     }
-}
\ 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/ConfigUpdater.java
index eaeb09e3..eb1a1a5f 100644
--- a/src/main/java/org/onap/dcae/configuration/ConfigLoader.java
+++ b/src/main/java/org/onap/dcae/configuration/ConfigUpdater.java
@@ -22,27 +22,23 @@ 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 {
+public class ConfigUpdater {
 
-    private static final Logger log = LoggerFactory.getLogger(ConfigLoader.class);
+    private static final Logger log = LoggerFactory.getLogger(ConfigUpdater.class);
     private final ConfigFilesFacade configFilesFacade;
-    private final CbsConfigResolver cbsConfigResolver;
     private final Runnable applicationRestarter;
     private boolean isApplicationRestartNeeded;
 
-    ConfigLoader(ConfigFilesFacade configFilesFacade, CbsConfigResolver cbsConfigResolver, Runnable applicationRestarter) {
+    public ConfigUpdater(ConfigFilesFacade configFilesFacade, Runnable applicationRestarter) {
         this.configFilesFacade = configFilesFacade;
-        this.cbsConfigResolver = cbsConfigResolver;
         this.applicationRestarter = applicationRestarter;
         this.isApplicationRestartNeeded = false;
     }
 
-    public synchronized void updateConfig() {
-        Option<JSONObject> appConfig = cbsConfigResolver.getAppConfig();
+    public synchronized void updateConfig(Option<JSONObject> appConfig) {
         appConfig.peek(this::handleUpdate).onEmpty(logSkipMessage());
     }
 
@@ -96,4 +92,4 @@ public class ConfigLoader {
             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/ConfigUpdaterFactory.java
index e5ea3c59..866bc20b 100644
--- a/src/main/java/org/onap/dcae/configuration/ConfigLoaderFactory.java
+++ b/src/main/java/org/onap/dcae/configuration/ConfigUpdaterFactory.java
@@ -21,17 +21,16 @@ 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 class ConfigUpdaterFactory {
 
-    public static ConfigLoader create(Path propertiesFile, Path dmaapConfigFile) {
+    private ConfigUpdaterFactory() {
+    }
+
+    public static ConfigUpdater create(Path propertiesFile, Path dmaapConfigFile) {
         ConfigFilesFacade configFilesFacade = new ConfigFilesFacade(propertiesFile, dmaapConfigFile);
-        CbsConfigResolver cbsConfigResolver = new CbsConfigResolverFactory().create();
-        return new ConfigLoader(
+        return new ConfigUpdater(
             configFilesFacade,
-            cbsConfigResolver,
-            VesApplication::restartApplication);
+                VesApplication::restartApplication);
     }
-}
\ No newline at end of file
+}
diff --git a/src/main/java/org/onap/dcae/configuration/ConfigurationHandler.java b/src/main/java/org/onap/dcae/configuration/ConfigurationHandler.java
new file mode 100644
index 00000000..ebdf0474
--- /dev/null
+++ b/src/main/java/org/onap/dcae/configuration/ConfigurationHandler.java
@@ -0,0 +1,109 @@
+/*
+ * ============LICENSE_START=======================================================
+ * VES Collector
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.s
+ * ================================================================================
+ * 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 com.google.gson.JsonObject;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.onap.dcae.configuration.cbs.CbsClientConfigurationProvider;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+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;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+
+/**
+ * ConfigurationHandler is responsible for receiving configuration updates from Consul.
+ * Any change made in the Consul will be reported as a notification.
+ */
+public class ConfigurationHandler {
+
+    private static Logger log = LoggerFactory.getLogger(ConfigurationHandler.class);
+    private static final String CONFIG_DICT = "config";
+
+    private final CbsClientConfigurationProvider cbsClientConfigurationProvider;
+    private final ConfigUpdater configUpdater;
+
+    /**
+     * Constructor
+     * @param cbsClientConfigurationProvider provides configuration to connect with Consul
+     * @param configUpdater for updating application configuration
+     */
+    public ConfigurationHandler(CbsClientConfigurationProvider cbsClientConfigurationProvider, ConfigUpdater configUpdater) {
+        this.cbsClientConfigurationProvider = cbsClientConfigurationProvider;
+        this.configUpdater = configUpdater;
+    }
+
+    /**
+     * Start listen for application configuration notifications with configuration changes
+     * @param interval defines period of time when notification can come
+     * @return {@link Disposable} handler to close Consul listener at the end
+     */
+    public Disposable startListen(Duration interval) {
+
+        log.info("Start listening for configuration from Consul ...");
+        log.info(String.format("Consul configuration will be fetched in %s period.", interval));
+
+        // Polling properties
+        final Duration initialDelay = Duration.ofSeconds(5);
+        final Duration period = interval;
+
+        final CbsRequest request = createCbsRequest();
+        final CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationProvider.get();
+
+        return createCbsClient(cbsClientConfiguration)
+                .flatMapMany(cbsClient -> cbsClient.updates(request, initialDelay, period))
+                .subscribe(
+                        this::handleConfigurationFromConsul,
+                        this::handleError
+                );
+    }
+
+    Mono<CbsClient> createCbsClient(CbsClientConfiguration cbsClientConfiguration) {
+        return CbsClientFactory.createCbsClient(cbsClientConfiguration);
+    }
+
+    void handleConfigurationFromConsul(JsonObject jsonObject) {
+        log.info("Configuration update from Consul {}", jsonObject);
+        if(jsonObject.has(CONFIG_DICT)) {
+            JsonObject config = jsonObject.getAsJsonObject(CONFIG_DICT);
+            JSONObject jObject = new JSONObject(config.toString());
+            configUpdater.updateConfig(Option.of(jObject));
+        } else {
+            throw new IllegalArgumentException(String.format("Invalid application configuration: %s ", jsonObject));
+        }
+    }
+
+    private void handleError(Throwable throwable) {
+        log.error("Unexpected error occurred during fetching configuration from Consul", throwable);
+    }
+
+    private CbsRequest createCbsRequest() {
+        RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
+        return CbsRequests.getAll(diagnosticContext);
+    }
+}
diff --git a/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolver.java b/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationProvider.java
index dc02131e..fc88197c 100644
--- a/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolver.java
+++ b/src/main/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationProvider.java
@@ -19,33 +19,58 @@
  */
 package org.onap.dcae.configuration.cbs;
 
+import org.jetbrains.annotations.NotNull;
 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 {
+/**
+ * CbsClientConfigurationProvider is used to provide production or dev configuration for CBS client.
+ */
+public class CbsClientConfigurationProvider {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(CbsClientConfigurationResolver.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(CbsClientConfigurationProvider.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";
+    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";
+    private static final String DEV_MODE_PROPERTY = "devMode";
+    private static final String CBS_PORT_PROPERTY = "cbsPort";
 
-    CbsClientConfiguration resolveCbsClientConfiguration() {
+    /**
+     * Returns configuration for CBS client.
+     * @return Production or dev configuration for CBS client, depends on application run arguments.
+     */
+    public CbsClientConfiguration get() {
         try {
-            return CbsClientConfiguration.fromEnvironment();
+            if (isDevModeEnabled()) {
+                return getDevConfiguration();
+            } else {
+                return CbsClientConfiguration.fromEnvironment();
+            }
         } catch (Exception e) {
-            LOGGER.warn("Failed resolving CBS client configuration from system environments: " + e);
+            LOGGER.warn(String.format("Failed resolving CBS client configuration from system environments: %s", e));
         }
-        LOGGER.info("Falling back to use default CBS client configuration properties");
         return getFallbackConfiguration();
     }
 
+    @NotNull
+    private ImmutableCbsClientConfiguration getDevConfiguration() {
+        return createCbsClientConfiguration(
+                DEFAULT_PROTOCOL, DEFAULT_HOSTNAME, DEFAULT_APP_NAME,
+                Integer.parseInt(System.getProperty(CBS_PORT_PROPERTY, String.valueOf(DEFAULT_PORT)))
+        );
+    }
+
+    private boolean isDevModeEnabled() {
+        return System.getProperty(DEV_MODE_PROPERTY) != null;
+    }
+
     private ImmutableCbsClientConfiguration getFallbackConfiguration() {
         LOGGER.info("Falling back to use default CBS client configuration");
-        return createCbsClientConfiguration(defaultProtocol, defaultHostname, defaultAppName, defaultPort);
+        return createCbsClientConfiguration(DEFAULT_PROTOCOL, DEFAULT_HOSTNAME, DEFAULT_APP_NAME, DEFAULT_PORT);
     }
 
     private ImmutableCbsClientConfiguration createCbsClientConfiguration(String protocol, String hostname,
@@ -57,4 +82,4 @@ class CbsClientConfigurationResolver {
             .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
deleted file mode 100644
index 09a96985..00000000
--- a/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolver.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*-
- * ============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 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 CbsClientConfigurationResolver cbsClientConfigurationResolver;
-    private final RequestDiagnosticContext diagnosticContext = RequestDiagnosticContext.create();
-    private final CbsRequest cbsConfigurationRequest = CbsRequests.getConfiguration(diagnosticContext);
-
-    CbsConfigResolver(CbsClientConfigurationResolver cbsClientConfigurationResolver) {
-        this.cbsClientConfigurationResolver = cbsClientConfigurationResolver;
-    }
-
-    public Option<JSONObject> getAppConfig() {
-        JsonObject emptyJson = new JsonObject();
-        CbsClientConfiguration cbsClientConfiguration = cbsClientConfigurationResolver.resolveCbsClientConfiguration();
-        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
deleted file mode 100644
index 5e42d9ec..00000000
--- a/src/main/java/org/onap/dcae/configuration/cbs/CbsConfigResolverFactory.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*-
- * ============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;
-
-public class CbsConfigResolverFactory {
-
-    public CbsConfigResolver create() {
-        CbsClientConfigurationResolver resolver = new CbsClientConfigurationResolver();
-        return new CbsConfigResolver(resolver);
-    }
-}
\ No newline at end of file
diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml
index e9a0f59c..3f915d02 100644
--- a/src/main/resources/log4j2.xml
+++ b/src/main/resources/log4j2.xml
@@ -87,6 +87,15 @@
   </Appenders>
 
   <Loggers>
+    <logger additivity="false" level="error" name="org.onap.dcaegen2.services.sdk">
+      <AppenderRef ref="ROL_CONSOLE"/>
+      <AppenderRef ref="EFILE"/>
+    </logger>
+
+    <logger additivity="false" level="error" name="org.onap.dcae.restapi">
+      <AppenderRef ref="ROL_CONSOLE"/>
+      <AppenderRef ref="EFILE"/>
+    </logger>
 
     <Logger additivity="false" level="info" name="org.onap.dcae.common.input">
       <AppenderRef ref="IFILE"/>
diff --git a/src/test/java/org/onap/dcae/ApplicationConfigurationListenerTest.java b/src/test/java/org/onap/dcae/ApplicationConfigurationListenerTest.java
new file mode 100644
index 00000000..49132f1e
--- /dev/null
+++ b/src/test/java/org/onap/dcae/ApplicationConfigurationListenerTest.java
@@ -0,0 +1,67 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * VES Collector
+ * ================================================================================
+ * 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;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dcae.configuration.ConfigurationHandler;
+
+import java.time.Duration;
+
+import static org.mockito.ArgumentMatchers.any;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ApplicationConfigurationListenerTest {
+
+    @Mock
+    private ConfigurationHandler configurationHandler;
+
+    @InjectMocks
+    @Spy
+    private ApplicationConfigurationListener applicationConfigurationListener;
+
+    @Test
+    public void shouldStopJobAndCloseConnectionWhenErrorOccurredDuringListenAtConsulChange() {
+
+       // given
+       Mockito.doThrow(new RuntimeException("Simulate exception")).when(configurationHandler).startListen(any());
+
+       // when
+       applicationConfigurationListener.run();
+
+       // then
+        Mockito.verify(applicationConfigurationListener).stopListeningForConfigurationUpdates(any());
+    }
+
+    @Test
+    public void shouldSendReloadAction() {
+
+        // when
+        applicationConfigurationListener.reload(Duration.ofMillis(1));
+
+        // then
+        Mockito.verify(applicationConfigurationListener).sendReloadAction();
+    }
+}
diff --git a/src/test/java/org/onap/dcae/configuration/CbsConfigurationHandlerTest.java b/src/test/java/org/onap/dcae/configuration/CbsConfigurationHandlerTest.java
new file mode 100644
index 00000000..7b1af1de
--- /dev/null
+++ b/src/test/java/org/onap/dcae/configuration/CbsConfigurationHandlerTest.java
@@ -0,0 +1,117 @@
+/*
+ * ============LICENSE_START=======================================================
+ * VES Collector
+ * ================================================================================
+ * Copyright (C) 2020 Nokia. All rights reserved.s
+ * ================================================================================
+ * 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 com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import io.vavr.control.Option;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dcae.configuration.cbs.CbsClientConfigurationProvider;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.CbsClientConfiguration;
+import reactor.core.Disposable;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class CbsConfigurationHandlerTest {
+
+    private static final String VES_CONFIG = "{\"collector.port\": 8081}";
+    private static final String VES_CONSUL_CONFIG = String.format("{\"config\":%s}", VES_CONFIG);
+
+    @Mock
+    private CbsClientConfigurationProvider cbsClientConfigurationProvider;
+    @Mock
+    private CbsClientConfiguration cbsClientConfiguration;
+    @Mock
+    private Mono<CbsClient> cbsClient;
+    @Mock
+    private ConfigUpdater configLoader;
+
+    @InjectMocks
+    @Spy
+    private ConfigurationHandler cbsConfigurationHandler;
+
+    @Test
+    public void shouldCreateCbsConfigurationHandler() {
+        // given
+
+        when(cbsConfigurationHandler.createCbsClient(cbsClientConfiguration)).thenReturn(cbsClient);
+        when(cbsClientConfigurationProvider.get()).thenReturn(cbsClientConfiguration);
+
+        // when
+        final Disposable handler = cbsConfigurationHandler.startListen(Duration.ofMinutes(5));
+
+        // then
+        assertThat(handler).isNotNull();
+    }
+
+    @Test
+    public void shouldUpdateAppConfigurationWhenConfigurationIsValid() {
+        // given
+        final JsonObject configuration = createConsulConfiguration(VES_CONSUL_CONFIG);
+
+        // when
+        this.cbsConfigurationHandler.handleConfigurationFromConsul(configuration);
+
+        // then
+        final ArgumentCaptor<Option<JSONObject>> acConfiguration = ArgumentCaptor.forClass(Option.class);
+        verify(configLoader).updateConfig(acConfiguration.capture());
+        assertThat(acConfiguration.getValue().get().toString()).hasToString(createJSONObject(VES_CONFIG));
+    }
+
+    @Test
+    public void shouldReportAnErrorWhenConsulReturnsEmptyConfiguration() {
+        // given
+        final JsonObject configuration = createConsulConfiguration("{}");
+
+        // when
+        assertThatThrownBy(() -> this.cbsConfigurationHandler.handleConfigurationFromConsul(configuration))
+                .isInstanceOf(IllegalArgumentException.class).hasMessageContaining(String.format("Invalid application configuration: %s ", "{}"));
+
+        // then
+        verify(configLoader, never()).updateConfig(any());
+    }
+
+    private String createJSONObject(String vesConfig) {
+        return new JSONObject(vesConfig).toString();
+    }
+
+    private JsonObject createConsulConfiguration(String vesConsulConfig) {
+        return new JsonParser().parse(vesConsulConfig).getAsJsonObject();
+    }
+
+
+}
diff --git a/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java b/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java
index 46d864a0..4d5fd921 100644
--- a/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java
+++ b/src/test/java/org/onap/dcae/configuration/ConfigLoaderTest.java
@@ -40,7 +40,6 @@ 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 {
@@ -53,14 +52,12 @@ public class ConfigLoaderTest {
     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;
+    private ConfigUpdater configLoader;
 
     @Mock
     private Runnable applicationRestarter;
@@ -73,28 +70,14 @@ public class ConfigLoaderTest {
     }
 
     @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);
+        Map<String, String> properties = HashMap.of(COLLECTOR_PORT, PORT_8080);
         mockVesInitialProperties(properties);
-        mockVesConfigInCbs(properties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(properties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, never()).writeProperties(any());
@@ -104,13 +87,13 @@ public class ConfigLoaderTest {
     @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);
+        Map<String, String> initialProperties = HashMap.of(COLLECTOR_PORT, PORT_8080);
+        Map<String, String> cbsProperties = HashMap.of(COLLECTOR_PORT, PORT_8081);
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -120,16 +103,16 @@ public class ConfigLoaderTest {
     @Test
     public void shouldUpdatePropertiesWhenVesKeysAreSubsetOfCbsKeysAndSubsetHasSameValues() {
         // given
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -138,16 +121,16 @@ public class ConfigLoaderTest {
 
     @Test
     public void shouldUpdatePropertiesWhenVesKeysAreSubsetOfCbsKeysAndSubsetHasDifferentValues() {
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8081,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -156,16 +139,16 @@ public class ConfigLoaderTest {
 
     @Test
     public void shouldNotUpdatePropertiesWhenCbsKeysAreSubsetOfVesKeysAndSubsetHasSameValues() {
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080);
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, never()).writeProperties(any());
@@ -174,16 +157,16 @@ public class ConfigLoaderTest {
 
     @Test
     public void shouldUpdatePropertiesWhenCbsKeysAreSubsetOfVesKeysAndSubsetHasDifferentValues() {
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8081);
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -192,18 +175,18 @@ public class ConfigLoaderTest {
 
     @Test
     public void shouldUpdatePropertiesWhenVesAndCbsKeySetsIntersectAndIntersectingKeysHaveSameValues() {
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_SCHEMA_FILE, SOME_SCHEMA
         );
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -212,18 +195,17 @@ public class ConfigLoaderTest {
 
     @Test
     public void shouldUpdatePropertiesWhenVesAndCbsKeySetsIntersectAndIntersectingKeysHaveDifferentValues() {
-        HashMap<String, String> initialProperties = HashMap.of(
+        Map<String, String> initialProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8080,
             COLLECTOR_KEYSTORE_FILE_LOCATION, SOME_PATH);
-        HashMap<String, String> cbsProperties = HashMap.of(
+        Map<String, String> cbsProperties = HashMap.of(
             COLLECTOR_PORT, PORT_8081,
             COLLECTOR_SCHEMA_FILE, SOME_SCHEMA
         );
         mockVesInitialProperties(initialProperties);
-        mockVesConfigInCbs(cbsProperties);
-
+        final Option<JSONObject> configuration = givenVesConfigInCbs(cbsProperties);
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, times(1)).writeProperties(cbsProperties);
@@ -236,10 +218,10 @@ public class ConfigLoaderTest {
         JSONObject emptyDmaapConfig = new JSONObject();
         JSONObject dmaapConfig = loadSampleDmaapConfig();
         mockVesInitialDmaapConfig(emptyDmaapConfig);
-        mockVesDmaapConfigInCbs(dmaapConfig);
+        final Option<JSONObject> configuration = givenVesDmaapConfigInCbs(dmaapConfig);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock).writeDMaaPConfiguration(argThat(dmaapConfig::similar));
@@ -251,10 +233,10 @@ public class ConfigLoaderTest {
         // given
         JSONObject dmaapConf = loadSampleDmaapConfig();
         mockVesInitialDmaapConfig(dmaapConf);
-        mockVesDmaapConfigInCbs(dmaapConf);
+        final Option<JSONObject> configuration = givenVesDmaapConfigInCbs(dmaapConf);
 
         // when
-        configLoader.updateConfig();
+        configLoader.updateConfig(configuration);
 
         // then
         verify(configFilesFacadeMock, never()).writeDMaaPConfiguration(any());
@@ -265,16 +247,16 @@ public class ConfigLoaderTest {
         when(configFilesFacadeMock.readDMaaPConfiguration()).thenReturn(Try.of(() -> dmaapConf));
     }
 
-    private void mockVesDmaapConfigInCbs(JSONObject dmaapConf) {
+    private Option<JSONObject> givenVesDmaapConfigInCbs(JSONObject dmaapConf) {
         JSONObject jsonObject = new JSONObject(f("{\"streams_publishes\": %s}}", dmaapConf));
-        when(cbsConfigResolverMock.getAppConfig()).thenReturn(Option.of(jsonObject));
+        return Option.of(jsonObject);
     }
 
-    private void mockVesConfigInCbs(HashMap<String, String> properties) {
-        when(cbsConfigResolverMock.getAppConfig()).thenReturn(Option.of(prepareConfigurationJson(properties)));
+    private Option<JSONObject> givenVesConfigInCbs(Map<String, String> properties) {
+        return Option.of(prepareConfigurationJson(properties));
     }
 
-    private void mockVesInitialProperties(HashMap<String, String> properties) {
+    private void mockVesInitialProperties(Map<String, String> properties) {
         when(configFilesFacadeMock.readCollectorProperties()).thenReturn(Try.of(() -> properties));
     }
 
@@ -291,4 +273,4 @@ public class ConfigLoaderTest {
         String jsonBody = f(template, customProperties);
         return new JSONObject(jsonBody);
     }
-}
\ No newline at end of file
+}
diff --git a/src/test/java/org/onap/dcae/configuration/ConfigLoaderFactoryTest.java b/src/test/java/org/onap/dcae/configuration/ConfigUpdaterFactoryTest.java
index e9421909..352fba1d 100644
--- a/src/test/java/org/onap/dcae/configuration/ConfigLoaderFactoryTest.java
+++ b/src/test/java/org/onap/dcae/configuration/ConfigUpdaterFactoryTest.java
@@ -24,20 +24,20 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.nio.file.Path;
 import org.junit.Test;
 
-public class ConfigLoaderFactoryTest {
+public class ConfigUpdaterFactoryTest {
 
     @Test
-    public void createsCbsConfigLoaderSuccessfully() {
+    public void createsCbsConfigUpdaterSuccessfully() {
         // given
         Path testPropertiesPath = Path.of("src/test/resources/testcollector.properties");
         Path testDmaapConfigPath = Path.of("src/test/resources/testParseDMaaPCredentialsGen2.json");
 
         // when
-        ConfigLoader configLoader = ConfigLoaderFactory.create(
+        ConfigUpdater configLoader = ConfigUpdaterFactory.create(
             testPropertiesPath,
             testDmaapConfigPath);
 
         // then
         assertThat(configLoader).isNotNull();
     }
-}
\ No newline at end of file
+}
diff --git a/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolverTest.java b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationProviderTest.java
index 57824595..b3c06d9c 100644
--- a/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationResolverTest.java
+++ b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientConfigurationProviderTest.java
@@ -25,7 +25,7 @@ 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 {
+public class CbsClientConfigurationProviderTest {
 
     private static final String DEFAULT_PROTOCOL = "http";
     private static final String DEFAULT_HOSTNAME = "config-binding-service";
@@ -36,7 +36,7 @@ public class CbsClientConfigurationResolverTest {
     @DisabledIfEnvironmentVariable(named = "CONFIG_BINDING_SERVICE", matches = ".+")
     public void shouldLoadDefaultConfigWhenEnvNotPresent() {
         // when
-        CbsClientConfiguration configuration = new CbsClientConfigurationResolver().resolveCbsClientConfiguration();
+        CbsClientConfiguration configuration = new CbsClientConfigurationProvider().get();
 
         // then
         assertThat(configuration.protocol()).isEqualTo(DEFAULT_PROTOCOL);
@@ -44,4 +44,4 @@ public class CbsClientConfigurationResolverTest {
         assertThat(configuration.port()).isEqualTo(DEFAULT_PORT);
         assertThat(configuration.appName()).isEqualTo(DEFAULT_APP_NAME);
     }
-}
\ No newline at end of file
+}
diff --git a/src/test/java/org/onap/dcae/configuration/cbs/CbsClientFactoryTest.java b/src/test/java/org/onap/dcae/configuration/cbs/CbsClientFactoryTest.java
deleted file mode 100644
index 4b17f68d..00000000
--- a/src/test/java/org/onap/dcae/configuration/cbs/CbsClientFactoryTest.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*-
- * ============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;
-
-public class CbsClientFactoryTest {
-
-    @Test
-    public void createsClientSuccessfully() {
-        // when
-        CbsConfigResolver cbsConfigResolver = new CbsConfigResolverFactory().create();
-
-        // 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
deleted file mode 100644
index 6f729997..00000000
--- a/src/test/java/org/onap/dcae/configuration/cbs/CbsConfigResolverTest.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*-
- * ============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.InjectMocks;
-import org.mockito.Mock;
-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;
-import static org.mockito.Mockito.when;
-
-@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());
-
-    @Mock
-    private CbsClientConfigurationResolver cbsClientConfigurationResolver;
-
-    @InjectMocks
-    private CbsConfigResolver cbsConfigResolver;
-
-    @Test
-    public void shouldFetchConfigurationFromCBS() {
-        // given
-        stubCBSToReturnAppConfig();
-        mockCbsClientConfiguration();
-
-        // when
-        JSONObject appConfig = cbsConfigResolver.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)));
-    }
-
-    private void mockCbsClientConfiguration() {
-        final int PORT = wireMockRule.port();
-        CbsClientConfiguration cbsClientConfiguration = ImmutableCbsClientConfiguration.builder()
-            .protocol(PROTOCOL)
-            .hostname(HOSTNAME)
-            .port(PORT)
-            .appName(APP_NAME)
-            .build();
-        when(cbsClientConfigurationResolver.resolveCbsClientConfiguration()).thenReturn(cbsClientConfiguration);
-    }
-}
\ No newline at end of file
diff --git a/version.properties b/version.properties
index 6460498f..6501a4a6 100644
--- a/version.properties
+++ b/version.properties
@@ -1,6 +1,6 @@
 major=1
 minor=7
-patch=8
+patch=9
 base_version=${major}.${minor}.${patch}
 release_version=${base_version}
 snapshot_version=${base_version}-SNAPSHOT