Cbs Client integration

Cbs Client is integrated to read configurations from consul

Issue-ID: DCAEGEN2-1595
Change-Id: Idb0ebd34eba077f9c1cb584abab4d8722b56f6c5
Signed-off-by: YongchaoWu <yongchao.wu@est.tech>

13 files changed, 244 insertions, 203 deletions

diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
index d9ca7871..f661dd0e 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfigTest.java
@@ -19,20 +19,17 @@ package org.onap.dcaegen2.collectors.datafile.configuration;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertTrue;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import com.google.common.base.Charsets;
 import com.google.common.io.Resources;
 import com.google.gson.JsonElement;
@@ -40,29 +37,28 @@ import com.google.gson.JsonIOException;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import com.google.gson.JsonSyntaxException;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
 import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.EnvProperties;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.http.configuration.ImmutableEnvProperties;
-import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.providers.CloudConfigurationProvider;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.api.CbsClient;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.EnvProperties;
+import org.onap.dcaegen2.services.sdk.rest.services.cbs.client.model.ImmutableEnvProperties;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapConsumerConfiguration;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapPublisherConfiguration;
-import reactor.core.Disposable;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -73,12 +69,14 @@ import reactor.test.StepVerifier;
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/9/18
  * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a>
  */
-class AppConfigTest {
+public class AppConfigTest {
 
-    private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
+    public static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
 
-    private static final DmaapConsumerConfiguration CORRECT_DMAAP_CONSUMER_CONFIG = //
+    public static final ImmutableDmaapConsumerConfiguration CORRECT_DMAAP_CONSUMER_CONFIG = //
         new ImmutableDmaapConsumerConfiguration.Builder() //
+            .endpointUrl(
+                "http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12")
             .timeoutMs(-1) //
             .dmaapHostName("message-router.onap.svc.cluster.local") //
             .dmaapUserName("admin") //
@@ -97,7 +95,7 @@ class AppConfigTest {
             .enableDmaapCertAuth(true) //
             .build();
 
-    private static final ConsumerConfiguration CORRECT_CONSUMER_CONFIG = ImmutableConsumerConfiguration.builder() //
+    public static final ConsumerConfiguration CORRECT_CONSUMER_CONFIG = ImmutableConsumerConfiguration.builder() //
         .topicUrl(
             "http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12")
         .trustStorePath("trustStorePath") //
@@ -120,7 +118,7 @@ class AppConfigTest {
             .passWord("password") //
             .build();
 
-    private static final FtpesConfig CORRECT_FTPES_CONFIGURATION = //
+    private static final ImmutableFtpesConfig CORRECT_FTPES_CONFIGURATION = //
         new ImmutableFtpesConfig.Builder() //
             .keyCert("/config/dfc.jks") //
             .keyPassword("secret") //
@@ -128,9 +126,9 @@ class AppConfigTest {
             .trustedCaPassword("secret") //
             .build();
 
-    private static final DmaapPublisherConfiguration CORRECT_DMAAP_PUBLISHER_CONFIG = //
+    private static final ImmutableDmaapPublisherConfiguration CORRECT_DMAAP_PUBLISHER_CONFIG = //
         new ImmutableDmaapPublisherConfiguration.Builder() //
-            .dmaapTopicName("/publish/1") //
+            .endpointUrl("https://message-router.onap.svc.cluster.local:3907/publish/1").dmaapTopicName("/publish/1") //
             .dmaapUserPassword("password") //
             .dmaapPortNumber(3907) //
             .dmaapProtocol("https") //
@@ -154,14 +152,14 @@ class AppConfigTest {
     }
 
     private AppConfig appConfigUnderTest;
-    private CloudConfigurationProvider cloudConfigurationProvider = mock(CloudConfigurationProvider.class);
     private final Map<String, String> context = MappedDiagnosticContext.initializeTraceContext();
+    CbsClient cbsClient = mock(CbsClient.class);
 
     @BeforeEach
-    public void setUp() {
+    void setUp() {
         appConfigUnderTest = spy(AppConfig.class);
-        appConfigUnderTest.setCloudConfigurationProvider(cloudConfigurationProvider);
         appConfigUnderTest.systemEnvironment = new Properties();
+
     }
 
     @Test
@@ -212,19 +210,14 @@ class AppConfigTest {
         // Given
         appConfigUnderTest.setFilepath("/temp.json");
 
-        ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
-
         // When
         appConfigUnderTest.loadConfigurationFromFile();
 
         // Then
-        assertTrue("Error message missing in log.",
-            logAppender.list.toString().contains("[WARN] Local configuration file not loaded: /temp.json"));
-        logAppender.stop();
-
         Assertions.assertNull(appConfigUnderTest.getDmaapConsumerConfiguration());
         assertThatThrownBy(() -> appConfigUnderTest.getPublisherConfiguration(CHANGE_IDENTIFIER))
             .hasMessageContaining("No PublishingConfiguration loaded, changeIdentifier: PM_MEAS_FILES");
+
         Assertions.assertNull(appConfigUnderTest.getFtpesConfiguration());
     }
 
@@ -264,32 +257,31 @@ class AppConfigTest {
     @Test
     public void whenPeriodicConfigRefreshNoEnvironmentVariables() {
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
-
-        Flux<AppConfig> task = appConfigUnderTest.createRefreshConfigurationTask(1L, context);
+        Flux<AppConfig> task = appConfigUnderTest.createRefreshTask(context);
 
         StepVerifier //
             .create(task) //
             .expectSubscription() //
-            .expectNextCount(0) //
-            .verifyComplete();
+            .verifyComplete(); //
 
         assertTrue(logAppender.list.toString().contains("$CONSUL_HOST environment has not been defined"));
     }
 
     @Test
     public void whenPeriodicConfigRefreshNoConsul() {
+        ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
+        EnvProperties props = properties();
+        doReturn(Mono.just(props)).when(appConfigUnderTest).getEnvironment(any(), any());
 
-        doReturn(Mono.just(properties())).when(appConfigUnderTest).readEnvironmentVariables(any(), any());
-        Mono<JsonObject> err = Mono.error(new IOException());
-        doReturn(err).when(cloudConfigurationProvider).callForServiceConfigurationReactive(any());
+        doReturn(Mono.just(cbsClient)).when(appConfigUnderTest).createCbsClient(props);
+        Flux<JsonObject> err = Flux.error(new IOException());
+        doReturn(err).when(cbsClient).updates(any(), any(), any());
 
-        ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(AppConfig.class);
-        Flux<AppConfig> task = appConfigUnderTest.createRefreshConfigurationTask(1L, context);
+        Flux<AppConfig> task = appConfigUnderTest.createRefreshTask(context);
 
         StepVerifier //
             .create(task) //
             .expectSubscription() //
-            .expectNextCount(0) //
             .verifyComplete();
 
         assertTrue(
@@ -298,13 +290,14 @@ class AppConfigTest {
 
     @Test
     public void whenPeriodicConfigRefreshSuccess() throws JsonIOException, JsonSyntaxException, IOException {
-        doReturn(Mono.just(properties())).when(appConfigUnderTest).readEnvironmentVariables(any(), any());
-
-        Mono<JsonObject> json = Mono.just(getJsonRootObject());
+        EnvProperties props = properties();
+        doReturn(Mono.just(props)).when(appConfigUnderTest).getEnvironment(any(), any());
+        doReturn(Mono.just(cbsClient)).when(appConfigUnderTest).createCbsClient(props);
 
-        doReturn(json, json).when(cloudConfigurationProvider).callForServiceConfigurationReactive(any());
+        Flux<JsonObject> json = Flux.just(getJsonRootObject());
+        doReturn(json).when(cbsClient).updates(any(), any(), any());
 
-        Flux<AppConfig> task = appConfigUnderTest.createRefreshConfigurationTask(1L, context);
+        Flux<AppConfig> task = appConfigUnderTest.createRefreshTask(context);
 
         StepVerifier //
             .create(task) //
@@ -317,14 +310,18 @@ class AppConfigTest {
 
     @Test
     public void whenPeriodicConfigRefreshSuccess2() throws JsonIOException, JsonSyntaxException, IOException {
-        doReturn(Mono.just(properties())).when(appConfigUnderTest).readEnvironmentVariables(any(), any());
+        EnvProperties props = properties();
+        doReturn(Mono.just(props)).when(appConfigUnderTest).getEnvironment(any(), any());
 
-        Mono<JsonObject> json = Mono.just(getJsonRootObject());
-        Mono<JsonObject> err = Mono.error(new IOException()); // no config entry created by the dmaap plugin
+        doReturn(Mono.just(cbsClient)).when(appConfigUnderTest).createCbsClient(props);
 
-        doReturn(json, err).when(cloudConfigurationProvider).callForServiceConfigurationReactive(any());
+        Flux<JsonObject> json = Flux.just(getJsonRootObject());
+        Flux<JsonObject> err = Flux.error(new IOException()); // no config entry created by the
+        // dmaap plugin
 
-        Flux<AppConfig> task = appConfigUnderTest.createRefreshConfigurationTask(1L, context);
+        doReturn(json, err).when(cbsClient).updates(any(), any(), any());
+
+        Flux<AppConfig> task = appConfigUnderTest.createRefreshTask(context);
 
         StepVerifier //
             .create(task) //
@@ -335,37 +332,6 @@ class AppConfigTest {
         Assertions.assertNotNull(appConfigUnderTest.getDmaapConsumerConfiguration());
     }
 
-    @Test
-    public void whenStopSuccess() {
-        Disposable disposableMock = mock(Disposable.class);
-        appConfigUnderTest.refreshConfigTask = disposableMock;
-
-        appConfigUnderTest.stop();
-
-        verify(disposableMock).dispose();
-        verifyNoMoreInteractions(disposableMock);
-        assertNull(appConfigUnderTest.refreshConfigTask);
-    }
-
-    @Test
-    public void whenNoPublisherConfigurationThrowException() throws DatafileTaskException {
-        appConfigUnderTest.publishingConfigurations = new HashMap<>();
-
-        DatafileTaskException exception = assertThrows(DatafileTaskException.class,
-            () -> appConfigUnderTest.getPublisherConfiguration(CHANGE_IDENTIFIER));
-        assertEquals("Cannot find getPublishingConfiguration for changeIdentifier: " + CHANGE_IDENTIFIER,
-            exception.getMessage());
-    }
-
-    @Test
-    public void whenFeedIsConfiguredReturnTrue() {
-        HashMap<String, PublisherConfiguration> publishingConfigs = new HashMap<>();
-        publishingConfigs.put(CHANGE_IDENTIFIER, null);
-        appConfigUnderTest.publishingConfigurations = publishingConfigs;
-
-        assertTrue(appConfigUnderTest.isFeedConfigured(CHANGE_IDENTIFIER));
-    }
-
     private JsonObject getJsonRootObject() throws JsonIOException, JsonSyntaxException, IOException {
         JsonObject rootObject = (new JsonParser()).parse(new InputStreamReader(getCorrectJson())).getAsJsonObject();
         return rootObject;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleControllerTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleControllerTest.java
index b630bd09..558eaf0e 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleControllerTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/ScheduleControllerTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import org.apache.commons.lang3.StringUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusControllerTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusControllerTest.java
index 9b8197f9..55c796ab 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusControllerTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/controllers/StatusControllerTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.doReturn;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import org.apache.commons.lang3.StringUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
index 5330a7f3..1c58650d 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClientTest.java
@@ -33,9 +33,11 @@ import com.jcraft.jsch.JSch;
 import com.jcraft.jsch.JSchException;
 import com.jcraft.jsch.Session;
 import com.jcraft.jsch.SftpException;
+
 import java.io.IOException;
 import java.nio.file.Paths;
 import java.util.Optional;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java
index 1e54d29d..499b2608 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/DmaapWebClientTest.java
@@ -25,8 +25,10 @@ import static org.mockito.Mockito.when;
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import java.net.URI;
 import java.net.URISyntaxException;
+
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
index 2e3245a4..cd18bfa2 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java
@@ -70,6 +70,7 @@ class JsonMessageParserTest {
     private static final String FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
     private static final String FILE_FORMAT_VERSION = "V10";
     private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
+    private static final String INCORRECT_CHANGE_IDENTIFIER = "INCORRECT_PM_MEAS_FILES";
     private static final String CHANGE_TYPE = "FileReady";
     private static final String INCORRECT_CHANGE_TYPE = "IncorrectFileReady";
     private static final String NOTIFICATION_FIELDS_VERSION = "1.0";
@@ -116,15 +117,14 @@ class JsonMessageParserTest {
             .files(files) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
             .getJsonObjectFromAnArray(jsonElement);
 
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNext(expectedMessage).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNext(expectedMessage).verifyComplete();
     }
 
     @Test
@@ -173,10 +173,11 @@ class JsonMessageParserTest {
         String messageString = "[" + parsedString + "," + parsedString + "]";
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
+        JsonElement jsonElement1 = new JsonParser().parse(messageString);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
             .getJsonObjectFromAnArray(jsonElement);
 
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement1)))
             .expectSubscription().expectNext(expectedMessage).expectNext(expectedMessage).verifyComplete();
     }
 
@@ -196,7 +197,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -204,8 +204,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue(logAppender.list.toString()
             .contains("[ERROR] VES event parsing. File information wrong. " + "Missing location."));
@@ -229,7 +229,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -237,8 +236,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
@@ -293,9 +292,10 @@ class JsonMessageParserTest {
         String parsedString = message.getParsed();
         String messageString = "[{\"event\":{}}," + parsedString + "]";
         JsonMessageParser jsonMessageParserUnderTest = new JsonMessageParser();
+        JsonElement jsonElement = new JsonParser().parse(messageString);
 
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNext(expectedMessage).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNext(expectedMessage).verifyComplete();
     }
 
     @Test
@@ -314,7 +314,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -322,8 +321,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectComplete().verify();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectComplete().verify();
 
         assertTrue("Error missing in log",
             logAppender.list.toString().contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
@@ -346,7 +345,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -354,8 +352,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
@@ -373,7 +371,6 @@ class JsonMessageParserTest {
             .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -381,8 +378,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue("Error missing in log",
             logAppender.list.toString().contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING
@@ -405,7 +402,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -413,8 +409,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
@@ -439,7 +435,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -447,8 +442,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).verifyComplete();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
@@ -506,15 +501,14 @@ class JsonMessageParserTest {
             .files(files) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
             .getJsonObjectFromAnArray(jsonElement);
 
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNext(expectedMessage).verifyComplete();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNext(expectedMessage).verifyComplete();
     }
 
     @Test
@@ -523,7 +517,6 @@ class JsonMessageParserTest {
             .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
             .build();
 
-        String incorrectMessageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -531,8 +524,8 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(incorrectMessageString)))
-            .expectSubscription().expectComplete().verify();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectComplete().verify();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
@@ -550,7 +543,7 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just("[{}]"))).expectSubscription()
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
             .expectComplete().verify();
 
         assertTrue("Error missing in log",
@@ -573,7 +566,6 @@ class JsonMessageParserTest {
             .addAdditionalField(additionalField) //
             .build();
 
-        String messageString = message.toString();
         String parsedString = message.getParsed();
         JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
         JsonElement jsonElement = new JsonParser().parse(parsedString);
@@ -581,12 +573,38 @@ class JsonMessageParserTest {
             .getJsonObjectFromAnArray(jsonElement);
 
         ListAppender<ILoggingEvent> logAppender = LoggingUtils.getLogListAppender(JsonMessageParser.class);
-        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(messageString)))
-            .expectSubscription().expectNextCount(0).expectComplete().verify();
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectNextCount(0).expectComplete().verify();
 
         assertTrue("Error missing in log",
             logAppender.list.toString()
                 .contains(ERROR_LOG_TAG + JsonMessageParser.ERROR_MSG_VES_EVENT_PARSING + " Change type is wrong: "
                     + INCORRECT_CHANGE_TYPE + " Expected: FileReady Message: " + message.getParsed()));
     }
+
+    @Test
+    void whenPassingCorrectJsonWithIncorrectChangeIdentifier_noFileData() {
+        AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() //
+            .name(PM_FILE_NAME) //
+            .location(LOCATION) //
+            .compression(GZIP_COMPRESSION) //
+            .fileFormatVersion(FILE_FORMAT_VERSION) //
+            .build();
+        JsonMessage message = new JsonMessage.JsonMessageBuilder() //
+            .eventName(NR_RADIO_ERICSSON_EVENT_NAME) //
+            .changeIdentifier(INCORRECT_CHANGE_IDENTIFIER) //
+            .changeType(CHANGE_TYPE) //
+            .notificationFieldsVersion(NOTIFICATION_FIELDS_VERSION) //
+            .addAdditionalField(additionalField) //
+            .build();
+
+        String parsedString = message.getParsed();
+        JsonMessageParser jsonMessageParserUnderTest = spy(new JsonMessageParser());
+        JsonElement jsonElement = new JsonParser().parse(parsedString);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject())).when(jsonMessageParserUnderTest)
+            .getJsonObjectFromAnArray(jsonElement);
+
+        StepVerifier.create(jsonMessageParserUnderTest.getMessagesFromJson(Mono.just(jsonElement))).expectSubscription()
+            .expectComplete().verify();
+    }
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java
index 1bea290f..a4319d37 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTest.java
@@ -20,21 +20,30 @@
 
 package org.onap.dcaegen2.collectors.datafile.tasks;
 
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
+import static org.onap.dcaegen2.collectors.datafile.configuration.AppConfigTest.CORRECT_CONSUMER_CONFIG;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
+import org.onap.dcaegen2.collectors.datafile.configuration.ConsumerConfiguration;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
 import org.onap.dcaegen2.collectors.datafile.ftp.Scheme;
 import org.onap.dcaegen2.collectors.datafile.model.FileData;
@@ -48,6 +57,7 @@ import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.service.JsonMessageParser;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage;
 import org.onap.dcaegen2.collectors.datafile.utils.JsonMessage.AdditionalField;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.ConsumerReactiveHttpClientFactory;
 import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consumer.DMaaPConsumerReactiveHttpClient;
 
 import reactor.core.publisher.Flux;
@@ -76,25 +86,36 @@ public class DMaaPMessageConsumerTest {
     private static final String GZIP_COMPRESSION = "gzip";
     private static final String MEAS_COLLECT_FILE_FORMAT_TYPE = "org.3GPP.32.435#measCollec";
     private static final String FILE_FORMAT_VERSION = "V10";
-    private static List<FilePublishInformation> listOfFilePublishInformation = new ArrayList<FilePublishInformation>();
+    private static List<FilePublishInformation> listOfFilePublishInformation = new ArrayList<>();
     private static final String CHANGE_IDENTIFIER = "PM_MEAS_FILES";
 
     private DMaaPConsumerReactiveHttpClient httpClientMock;
 
     private DMaaPMessageConsumer messageConsumer;
     private static String ftpesMessageString;
+    private static JsonElement ftpesMessageJson;
     private static FileData ftpesFileData;
     private static FileReadyMessage expectedFtpesMessage;
 
     private static String sftpMessageString;
+    private static JsonElement sftpMessageJson;
     private static FileData sftpFileData;
     private static FileReadyMessage expectedSftpMessage;
 
+    private static AppConfig appConfig;
+    private static ConsumerConfiguration dmaapConsumerConfiguration;
+
     /**
      * Sets up data for the test.
      */
     @BeforeAll
     public static void setUp() {
+
+        appConfig = mock(AppConfig.class);
+        dmaapConsumerConfiguration = CORRECT_CONSUMER_CONFIG;
+
+        JsonParser jsonParser = new JsonParser();
+
         AdditionalField ftpesAdditionalField = new JsonMessage.AdditionalFieldBuilder() //
             .location(FTPES_LOCATION) //
             .compression(GZIP_COMPRESSION) //
@@ -111,6 +132,8 @@ public class DMaaPMessageConsumerTest {
             .build();
 
         ftpesMessageString = ftpesJsonMessage.toString();
+        ftpesMessageJson = jsonParser.parse(ftpesMessageString);
+
         MessageMetaData messageMetaData = ImmutableMessageMetaData.builder() //
             .productName(PRODUCT_NAME) //
             .vendorName(VENDOR_NAME) //
@@ -151,6 +174,7 @@ public class DMaaPMessageConsumerTest {
             .addAdditionalField(sftpAdditionalField) //
             .build();
         sftpMessageString = sftpJsonMessage.toString();
+        sftpMessageJson = jsonParser.parse(sftpMessageString);
         sftpFileData = ImmutableFileData.builder() //
             .name(PM_FILE_NAME) //
             .location(SFTP_LOCATION) //
@@ -188,54 +212,62 @@ public class DMaaPMessageConsumerTest {
 
     @Test
     public void whenPassedObjectDoesntFit_ThrowsDatafileTaskException() {
-        prepareMocksForDmaapConsumer("", null);
+        prepareMocksForDmaapConsumer(Optional.empty(), null);
 
         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
             .expectSubscription() //
             .expectError(DatafileTaskException.class) //
             .verify();
 
-        verify(httpClientMock, times(1)).getDMaaPConsumerResponse();
+        verify(httpClientMock, times(1)).getDMaaPConsumerResponse(Optional.empty());
     }
 
     @Test
     public void whenFtpes_ReturnsCorrectResponse() throws DatafileTaskException {
-        prepareMocksForDmaapConsumer(ftpesMessageString, expectedFtpesMessage);
+        prepareMocksForDmaapConsumer(Optional.of(ftpesMessageJson), expectedFtpesMessage);
 
         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
             .expectNext(expectedFtpesMessage) //
             .verifyComplete();
 
-        verify(httpClientMock, times(1)).getDMaaPConsumerResponse();
+        verify(httpClientMock, times(1)).getDMaaPConsumerResponse(Optional.empty());
         verifyNoMoreInteractions(httpClientMock);
     }
 
     @Test
     public void whenSftp_ReturnsCorrectResponse() throws DatafileTaskException {
-        prepareMocksForDmaapConsumer(sftpMessageString, expectedSftpMessage);
+        prepareMocksForDmaapConsumer(Optional.of(sftpMessageJson), expectedSftpMessage);
 
         StepVerifier.create(messageConsumer.getMessageRouterResponse()) //
             .expectNext(expectedSftpMessage) //
             .verifyComplete();
 
-        verify(httpClientMock, times(1)).getDMaaPConsumerResponse();
+        verify(httpClientMock, times(1)).getDMaaPConsumerResponse(Optional.empty());
         verifyNoMoreInteractions(httpClientMock);
     }
 
-    private void prepareMocksForDmaapConsumer(String message, FileReadyMessage fileReadyMessageAfterConsume) {
-        Mono<String> messageAsMono = Mono.just(message);
+    private void prepareMocksForDmaapConsumer(Optional<JsonElement> message,
+        FileReadyMessage fileReadyMessageAfterConsume) {
+        Mono<JsonElement> messageAsMono = message.isPresent() ? Mono.just(message.get()) : Mono.empty();
         JsonMessageParser jsonMessageParserMock = mock(JsonMessageParser.class);
         httpClientMock = mock(DMaaPConsumerReactiveHttpClient.class);
-        when(httpClientMock.getDMaaPConsumerResponse()).thenReturn(messageAsMono);
+        when(httpClientMock.getDMaaPConsumerResponse(Optional.empty())).thenReturn(messageAsMono);
+        when(appConfig.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
+        ConsumerReactiveHttpClientFactory httpClientFactory = mock(ConsumerReactiveHttpClientFactory.class);
+        try {
+            doReturn(httpClientMock).when(httpClientFactory).create(dmaapConsumerConfiguration.toDmaap());
+        } catch (DatafileTaskException e) {
+            e.printStackTrace();
+        }
 
-        if (!message.isEmpty()) {
-            when(jsonMessageParserMock.getMessagesFromJson(messageAsMono))
-                .thenReturn(Flux.just(fileReadyMessageAfterConsume));
+        if (message.isPresent()) {
+            when(jsonMessageParserMock.getMessagesFromJson(any())).thenReturn(Flux.just(fileReadyMessageAfterConsume));
         } else {
-            when(jsonMessageParserMock.getMessagesFromJson(messageAsMono))
+            when(jsonMessageParserMock.getMessagesFromJson(any()))
                 .thenReturn(Flux.error(new DatafileTaskException("problemas")));
         }
 
-        messageConsumer = spy(new DMaaPMessageConsumer(httpClientMock, jsonMessageParserMock));
+        messageConsumer = spy(new DMaaPMessageConsumer(appConfig, jsonMessageParserMock, httpClientFactory));
     }
+
 }
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
index ddc279c2..fb369174 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java
@@ -29,6 +29,7 @@ import static org.mockito.Mockito.when;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import java.io.File;
 import java.net.URI;
 import java.nio.file.Path;
@@ -38,6 +39,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+
 import org.apache.http.Header;
 import org.apache.http.HttpResponse;
 import org.apache.http.StatusLine;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
index 93f20077..1ab97d4a 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java
@@ -32,6 +32,7 @@ import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
+
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
index 5a8d962f..a0096b77 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java
@@ -37,6 +37,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.time.Instant;
@@ -44,6 +45,7 @@ import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+
 import org.apache.commons.lang3.StringUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -65,6 +67,7 @@ import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData;
 import org.onap.dcaegen2.collectors.datafile.utils.LoggingUtils;
 import org.onap.dcaegen2.services.sdk.rest.services.model.logging.MdcVariables;
 import org.slf4j.MDC;
+
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/LoggingUtils.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/LoggingUtils.java
index 68f3582f..cfcb7bf9 100644
--- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/LoggingUtils.java
+++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/utils/LoggingUtils.java
@@ -24,6 +24,7 @@ import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.Logger;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.read.ListAppender;
+
 import org.slf4j.LoggerFactory;
 
 public class LoggingUtils {
diff --git a/datafile-app-server/src/test/resources/datafile_endpoints_test.json b/datafile-app-server/src/test/resources/datafile_endpoints_test.json
index 4d4d00ab..0157c7d2 100644
--- a/datafile-app-server/src/test/resources/datafile_endpoints_test.json
+++ b/datafile-app-server/src/test/resources/datafile_endpoints_test.json
@@ -1,31 +1,35 @@
 {
-   "dmaap.ftpesConfig.keyCert":"/config/dfc.jks",
-   "dmaap.ftpesConfig.keyPassword":"secret",
-   "dmaap.ftpesConfig.trustedCa":"config/ftp.jks",
-   "dmaap.ftpesConfig.trustedCaPassword":"secret",
-   "dmaap.security.trustStorePath":"trustStorePath",
-   "dmaap.security.trustStorePasswordPath":"trustStorePasswordPath",
-   "dmaap.security.keyStorePath":"keyStorePath",
-   "dmaap.security.keyStorePasswordPath":"keyStorePasswordPath",
-   "dmaap.security.enableDmaapCertAuth":"true",
-   "dmaap.dmaapProducerConfiguration": {
-         "changeIdentifier":"PM_MEAS_FILES",
-         "feedName":"feed00"
-   },
-   "streams_subscribes":{
-      "dmaap_subscriber":{
-         "dmmap_info":{
-            "topic_url":"http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12"
-         },
-         "type":"message_router"
+  "config": {
+    "//description": "This file is only used for testing purposes",
+    "dmaap.ftpesConfig.keyCert": "/config/dfc.jks",
+    "dmaap.ftpesConfig.keyPassword": "secret",
+    "dmaap.ftpesConfig.trustedCa": "config/ftp.jks",
+    "dmaap.ftpesConfig.trustedCaPassword": "secret",
+    "dmaap.security.trustStorePath": "trustStorePath",
+    "dmaap.security.trustStorePasswordPath": "trustStorePasswordPath",
+    "dmaap.security.keyStorePath": "keyStorePath",
+    "dmaap.security.keyStorePasswordPath": "keyStorePasswordPath",
+    "dmaap.security.enableDmaapCertAuth": "true",
+    "streams_publishes": {
+      "PM_MEAS_FILES": {
+        "type": "data_router",
+        "dmaap_info": {
+          "username": "user",
+          "log_url": "https://dmaap.example.com/feedlog/972",
+          "publish_url": "https://message-router.onap.svc.cluster.local:3907/publish/1",
+          "location": "loc00",
+          "password": "password",
+          "publisher_id": "972.360gm"
+        }
       }
-   },
-   "feed00":{
-      "username":"user",
-      "log_url":"https://dmaap.example.com/feedlog/972",
-      "publish_url":"https://message-router.onap.svc.cluster.local:3907/publish/1",
-      "location":"loc00",
-      "password":"password",
-      "publisher_id":"972.360gm"
-   }
-}
+    },
+    "streams_subscribes": {
+      "dmaap_subscriber": {
+        "dmaap_info": {
+          "topic_url": "http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12"
+        },
+        "type": "message_router"
+      }
+    }
+  }
+}
\ No newline at end of file
diff --git a/datafile-app-server/src/test/resources/datafile_endpoints_test_2producers.json b/datafile-app-server/src/test/resources/datafile_endpoints_test_2producers.json
index a7e2497d..61b324ce 100644
--- a/datafile-app-server/src/test/resources/datafile_endpoints_test_2producers.json
+++ b/datafile-app-server/src/test/resources/datafile_endpoints_test_2producers.json
@@ -1,49 +1,57 @@
 {
-   "dmaap.ftpesConfig.keyCert":"/config/dfc.jks",
-   "dmaap.ftpesConfig.keyPassword":"secret",
-   "dmaap.ftpesConfig.trustedCa":"config/ftp.jks",
-   "dmaap.ftpesConfig.trustedCaPassword":"secret",
-   "dmaap.security.trustStorePath":"trustStorePath",
-   "dmaap.security.trustStorePasswordPath":"trustStorePasswordPath",
-   "dmaap.security.keyStorePath":"keyStorePath",
-   "dmaap.security.keyStorePasswordPath":"keyStorePasswordPath",
-   "dmaap.security.enableDmaapCertAuth":"true",
-   "dmaap.dmaapProducerConfiguration":[
-      {
-         "changeIdentifier":"PM_MEAS_FILES",
-         "feedName":"feed00"
+  "config": {
+    "//description": "This file is only used for testing purposes",
+    "dmaap.ftpesConfig.keyCert": "/config/dfc.jks",
+    "dmaap.ftpesConfig.keyPassword": "secret",
+    "dmaap.ftpesConfig.trustedCa": "config/ftp.jks",
+    "dmaap.ftpesConfig.trustedCaPassword": "secret",
+    "dmaap.security.trustStorePath": "trustStorePath",
+    "dmaap.security.trustStorePasswordPath": "trustStorePasswordPath",
+    "dmaap.security.keyStorePath": "keyStorePath",
+    "dmaap.security.keyStorePasswordPath": "keyStorePasswordPath",
+    "dmaap.security.enableDmaapCertAuth": "true",
+    "streams_publishes": {
+      "PM_MEAS_FILES": {
+        "type": "data_router",
+        "dmaap_info": {
+          "username": "CYE9fl40",
+          "location": "loc00",
+          "log_url": "https://dmaap-dr-prov/feedlog/4",
+          "publisher_id": "4.307dw",
+          "password": "izBJD8nLjawq0HMG",
+          "publish_url": "https://dmaap-dr-prov/publish/4"
+        }
       },
-      {
-         "changeIdentifier":"XX_FILES",
-         "feedName":"feed01"
+      "XX_FILES": {
+        "type": "data_router",
+        "dmaap_info": {
+          "username": "user",
+          "log_url": "feed01::log_url",
+          "publish_url": "feed01::publish_url",
+          "location": "loc00",
+          "password": "",
+          "publisher_id": ""
+        }
       },
-      {
-         "changeIdentifier":"YY_FILES",
-         "feedName":"feed01"
+      "YY_FILES": {
+        "type": "data_router",
+        "dmaap_info": {
+          "username": "user",
+          "log_url": "feed01::log_url",
+          "publish_url": "feed01::publish_url",
+          "location": "loc00",
+          "password": "",
+          "publisher_id": ""
+        }
       }
-   ],
-   "streams_subscribes":{
-      "dmaap_subscriber":{
-         "dmmap_info":{
-            "topic_url":"http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12"
-         },
-         "type":"message_router"
+    },
+    "streams_subscribes": {
+      "dmaap_subscriber": {
+        "dmaap_info": {
+          "topic_url": "http://admin:admin@message-router.onap.svc.cluster.local:2222/events/unauthenticated.VES_NOTIFICATION_OUTPUT/OpenDcae-c12/C12"
+        },
+        "type": "message_router"
       }
-   },
-   "feed00":{
-      "username":"user",
-      "log_url":"https://dmaap.example.com/feedlog/972",
-      "publish_url":"https://message-router.onap.svc.cluster.local:3907/publish/1",
-      "location":"loc00",
-      "password":"password",
-      "publisher_id":"972.360gm"
-   },
-   "feed01":{
-      "username":"user",
-      "log_url":"feed01::log_url",
-      "publish_url":"feed01::publish_url",
-      "location":"loc00",
-      "password":"",
-      "publisher_id":""
-   }
-}
+    }
+  }
+}
\ No newline at end of file