PRH DMaaP objects batching

*Getting collection of object
in one request
*Refator the workflow
in the old implementation

Change-Id: I4fdbf4bd8ae70cd78dbf5c3c441ba01c28e6ce4f
Issue-ID: DCAEGEN2-834
Signed-off-by: wasala <przemyslaw.wasala@nokia.com>

2 files changed, 295 insertions, 248 deletions

diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParserTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParserTest.java
index 225d46ee..01ce7419 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParserTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParserTest.java
@@ -28,7 +28,6 @@ import java.util.Optional;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
-import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
 import reactor.core.publisher.Mono;
@@ -43,36 +42,36 @@ class DmaapConsumerJsonParserTest {
     void whenPassingCorrectJson_validationNotThrowingAnException() {
         //given
         String message = "[{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"oamV4IpAddress\": \"10.16.123.234\","
-                + " \"softwareVersion\": \"v4.5.0.1\","
-                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}]";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}]";
 
         String parsed = "{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"oamV4IpAddress\": \"10.16.123.234\","
-                + " \"softwareVersion\": \"v4.5.0.1\","
-                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}";
 
         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
             .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
@@ -83,7 +82,7 @@ class DmaapConsumerJsonParserTest {
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser
-            .getJsonObject(Mono.just((message))).block();
+            .getJsonObject(Mono.just((message))).blockFirst();
         //then
         Assertions.assertNotNull(consumerDmaapModel);
         Assertions.assertEquals(expectedObject, consumerDmaapModel);
@@ -93,34 +92,34 @@ class DmaapConsumerJsonParserTest {
     void whenPassingCorrectJsonWithoutIpv4_validationNotThrowingAnException() {
         //given
         String message = "[{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\","
-                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}]";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}]";
 
         String parsed = "{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\","
-                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}";
 
         //when
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
@@ -129,7 +128,7 @@ class DmaapConsumerJsonParserTest {
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         dmaapConsumerJsonParser.getJsonObject(Mono.just((message)));
         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
-            .block();
+            .blockFirst();
         //then
         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("")
             .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
@@ -142,34 +141,34 @@ class DmaapConsumerJsonParserTest {
     void whenPassingCorrectJsonWithoutIpv6_validationNotThrowingAnException() {
         //given
         String message = "[{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"oamV4IpAddress\": \"10.16.123.234\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
-                + "}}}]";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}]";
 
         String parsed = "{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"oamV4IpAddress\": \"10.16.123.234\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
-                + "}}}";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}";
 
         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234").ipv6("")
             .correlationId("NOKQTFCOC540002E").build();
@@ -179,55 +178,112 @@ class DmaapConsumerJsonParserTest {
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
-            .block();
+            .blockFirst();
         //then
         Assertions.assertNotNull(consumerDmaapModel);
         Assertions.assertEquals(expectedObject, consumerDmaapModel);
     }
 
     @Test
-    void whenPassingCorrectJsonWithoutIpv4andIpv6_validationThrowingAnException() {
+    void whenPassingCorrectJsonWithoutIpv4andIpv6_validationAddingAnException() {
         //given
         String message = "[{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
-                + "}}}]";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}]";
 
         String parsed = "{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
-                + "}}}";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}";
 
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
         JsonElement jsonElement = new JsonParser().parse(parsed);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+            .expectSubscription().thenRequest(1).verifyComplete();
 
     }
 
     @Test
-    void whenPassingJsonWithoutMandatoryHeaderInformation_validationThrowingAnException() {
+    void whenPassingJsonWithoutMandatoryHeaderInformation_validationAddingAnException() {
         String parsed = "{\"event\": {"
+            + "\"commonEventHeader\": {},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}";
+
+        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
+        JsonElement jsonElement = new JsonParser().parse(parsed);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+        String incorrectMessage = "[{\"event\": {"
+            + "\"commonEventHeader\": {},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\""
+            + "}}}]";
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessage)))
+            .expectSubscription().thenRequest(1).verifyComplete();
+    }
+
+    @Test
+    void whenPassingJsonWithoutSourceName_validationAddingAnException() {
+        String parsed = "{\"event\": {"
+            + "\"commonEventHeader\": {},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}";
+
+        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
+        JsonElement jsonElement = new JsonParser().parse(parsed);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+        String jsonWithoutSourceName =
+            "[{\"event\": {"
                 + "\"commonEventHeader\": {},"
                 + "\"pnfRegistrationFields\": {"
                 + " \"unitType\": \"AirScale\","
@@ -238,15 +294,41 @@ class DmaapConsumerJsonParserTest {
                 + " \"lastServiceDate\": \"1535014037024\","
                 + " \"unitFamily\": \"BBU\","
                 + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
+                + " \"softwareVersion\": \"v4.5.0.1\","
+                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+                + "}}}]";
+        StepVerifier
+            .create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutSourceName)))
+            .expectSubscription().thenRequest(1)
+            .verifyComplete();
+    }
+
+    @Test
+    void whenPassingJsonWithoutIpInformation_validationAddingAnException() {
+        String parsed =
+            "{\"event\": {"
+                + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
+                + "\"pnfRegistrationFields\": {"
+                + " \"unitType\": \"AirScale\","
+                + " \"serialNumber\": \"QTFCOC540002E\","
+                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+                + " \"manufactureDate\": \"1535014037024\","
+                + " \"modelNumber\": \"7BEA\",\n"
+                + " \"lastServiceDate\": \"1535014037024\","
+                + " \"unitFamily\": \"BBU\","
+                + " \"vendorName\": \"Nokia\","
+                + " \"softwareVersion\": \"v4.5.0.1\","
+                + " \"oamV4IpAddress\": \"\","
+                + " \"oamV6IpAddress\": \"\""
                 + "}}}";
 
         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
         JsonElement jsonElement = new JsonParser().parse(parsed);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String incorrectMessage = "[{\"event\": {"
-                + "\"commonEventHeader\": {},"
+        String jsonWithoutIpInformation =
+            "[{\"event\": {"
+                + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
                 + "\"pnfRegistrationFields\": {"
                 + " \"unitType\": \"AirScale\","
                 + " \"serialNumber\": \"QTFCOC540002E\","
@@ -256,16 +338,19 @@ class DmaapConsumerJsonParserTest {
                 + " \"lastServiceDate\": \"1535014037024\","
                 + " \"unitFamily\": \"BBU\","
                 + " \"vendorName\": \"Nokia\","
-                + " \"softwareVersion\": \"v4.5.0.1\""
+                + " \"softwareVersion\": \"v4.5.0.1\","
+                + " \"oamV4IpAddress\": \"\","
+                + " \"oamV6IpAddress\": \"\""
                 + "}}}]";
-        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessage)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
+            .expectSubscription().thenRequest(1).verifyComplete();
     }
 
     @Test
-    void whenPassingJsonWithoutSourceName_validationThrowingAnException() {
-        String parsed = "{\"event\": {"
-                + "\"commonEventHeader\": {},"
+    void whenPassingJsonWithoutSourceNameValue_validationAddingAnException() {
+        String parsed =
+            "{\"event\": {"
+                + "\"commonEventHeader\": {\"sourceName\": \"\"},"
                 + "\"pnfRegistrationFields\": {"
                 + " \"unitType\": \"AirScale\","
                 + " \"serialNumber\": \"QTFCOC540002E\","
@@ -276,6 +361,7 @@ class DmaapConsumerJsonParserTest {
                 + " \"unitFamily\": \"BBU\","
                 + " \"vendorName\": \"Nokia\","
                 + " \"softwareVersion\": \"v4.5.0.1\","
+                + " \"oamV4IpAddress\": \"10.16.123.234\","
                 + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
                 + "}}}";
 
@@ -283,109 +369,69 @@ class DmaapConsumerJsonParserTest {
         JsonElement jsonElement = new JsonParser().parse(parsed);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String jsonWithoutSourceName =
-                "[{\"event\": {"
-                        + "\"commonEventHeader\": {},"
-                        + "\"pnfRegistrationFields\": {"
-                        + " \"unitType\": \"AirScale\","
-                        + " \"serialNumber\": \"QTFCOC540002E\","
-                        + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                        + " \"manufactureDate\": \"1535014037024\","
-                        + " \"modelNumber\": \"7BEA\",\n"
-                        + " \"lastServiceDate\": \"1535014037024\","
-                        + " \"unitFamily\": \"BBU\","
-                        + " \"vendorName\": \"Nokia\","
-                        + " \"softwareVersion\": \"v4.5.0.1\","
-                        + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                        + "}}}]";
-        StepVerifier
-            .create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutSourceName)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
-    }
-
-    @Test
-    void whenPassingJsonWithoutIpInformation_validationThrowingAnException() {
-        String parsed =
-                "{\"event\": {"
-                        + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
-                        + "\"pnfRegistrationFields\": {"
-                        + " \"unitType\": \"AirScale\","
-                        + " \"serialNumber\": \"QTFCOC540002E\","
-                        + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                        + " \"manufactureDate\": \"1535014037024\","
-                        + " \"modelNumber\": \"7BEA\",\n"
-                        + " \"lastServiceDate\": \"1535014037024\","
-                        + " \"unitFamily\": \"BBU\","
-                        + " \"vendorName\": \"Nokia\","
-                        + " \"softwareVersion\": \"v4.5.0.1\","
-                        + " \"oamV4IpAddress\": \"\","
-                        + " \"oamV6IpAddress\": \"\""
-                        + "}}}";
-
-        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         String jsonWithoutIpInformation =
-                "[{\"event\": {"
-                        + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
-                        + "\"pnfRegistrationFields\": {"
-                        + " \"unitType\": \"AirScale\","
-                        + " \"serialNumber\": \"QTFCOC540002E\","
-                        + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                        + " \"manufactureDate\": \"1535014037024\","
-                        + " \"modelNumber\": \"7BEA\",\n"
-                        + " \"lastServiceDate\": \"1535014037024\","
-                        + " \"unitFamily\": \"BBU\","
-                        + " \"vendorName\": \"Nokia\","
-                        + " \"softwareVersion\": \"v4.5.0.1\","
-                        + " \"oamV4IpAddress\": \"\","
-                        + " \"oamV6IpAddress\": \"\""
-                        + "}}}]";
+            "[{\"event\": {"
+                + "\"commonEventHeader\": {\"sourceName\": \"\"},"
+                + "\"pnfRegistrationFields\": {"
+                + " \"unitType\": \"AirScale\","
+                + " \"serialNumber\": \"QTFCOC540002E\","
+                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+                + " \"manufactureDate\": \"1535014037024\","
+                + " \"modelNumber\": \"7BEA\",\n"
+                + " \"lastServiceDate\": \"1535014037024\","
+                + " \"unitFamily\": \"BBU\","
+                + " \"vendorName\": \"Nokia\","
+                + " \"softwareVersion\": \"v4.5.0.1\","
+                + " \"oamV4IpAddress\": \"10.16.123.234\","
+                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+                + "}}}]";
         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
-            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+            .expectSubscription().thenRequest(1).verifyComplete();
     }
 
     @Test
-    void whenPassingJsonWithoutSourceNameValue_validationThrowingAnException() {
-        String parsed =
-                "{\"event\": {"
-                        + "\"commonEventHeader\": {\"sourceName\": \"\"},"
-                        + "\"pnfRegistrationFields\": {"
-                        + " \"unitType\": \"AirScale\","
-                        + " \"serialNumber\": \"QTFCOC540002E\","
-                        + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                        + " \"manufactureDate\": \"1535014037024\","
-                        + " \"modelNumber\": \"7BEA\",\n"
-                        + " \"lastServiceDate\": \"1535014037024\","
-                        + " \"unitFamily\": \"BBU\","
-                        + " \"vendorName\": \"Nokia\","
-                        + " \"softwareVersion\": \"v4.5.0.1\","
-                        + " \"oamV4IpAddress\": \"10.16.123.234\","
-                        + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                        + "}}}";
+    void whenPassingCorrectJsoArraynWithoutIpv4_validationNotThrowingAnException() {
+        //given
+        String message = "[{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}},"
+            + "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}"
+            + "]";
 
-        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-                .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
-        String jsonWithoutIpInformation =
-                "[{\"event\": {"
-                        + "\"commonEventHeader\": {\"sourceName\": \"\"},"
-                        + "\"pnfRegistrationFields\": {"
-                        + " \"unitType\": \"AirScale\","
-                        + " \"serialNumber\": \"QTFCOC540002E\","
-                        + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                        + " \"manufactureDate\": \"1535014037024\","
-                        + " \"modelNumber\": \"7BEA\",\n"
-                        + " \"lastServiceDate\": \"1535014037024\","
-                        + " \"unitFamily\": \"BBU\","
-                        + " \"vendorName\": \"Nokia\","
-                        + " \"softwareVersion\": \"v4.5.0.1\","
-                        + " \"oamV4IpAddress\": \"10.16.123.234\","
-                        + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                        + "}}}]";
-        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
-                .expectSubscription().expectError(DmaapNotFoundException.class).verify();
+        ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
+            .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
+            .correlationId("NOKQTFCOC540002E").build();
+        //when
+        DmaapConsumerJsonParser dmaapConsumerJsonParser = new DmaapConsumerJsonParser();
+
+        //then
+        StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
+            .expectSubscription().expectNext(expectedObject).expectNext(expectedObject).verifyComplete();
     }
 }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
index c128fb95..689a732c 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImplTest.java
@@ -42,6 +42,7 @@ import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
 import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -71,36 +72,36 @@ class DmaapConsumerTaskImplTest {
         appConfig = mock(AppConfig.class);
 
         message = "[{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                    + " \"unitType\": \"AirScale\","
-                    + " \"serialNumber\": \"QTFCOC540002E\","
-                    + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                    + " \"manufactureDate\": \"1535014037024\","
-                    + " \"modelNumber\": \"7BEA\",\n"
-                    + " \"lastServiceDate\": \"1535014037024\","
-                    + " \"unitFamily\": \"BBU\","
-                    + " \"vendorName\": \"Nokia\","
-                    + " \"oamV4IpAddress\": \"10.16.123.234\","
-                    + " \"softwareVersion\": \"v4.5.0.1\","
-                    + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}]";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}]";
 
         parsed = "{\"event\": {"
-                + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
-                + "\"pnfRegistrationFields\": {"
-                + " \"unitType\": \"AirScale\","
-                + " \"serialNumber\": \"QTFCOC540002E\","
-                + " \"pnfRegistrationFieldsVersion\": \"2.0\","
-                + " \"manufactureDate\": \"1535014037024\","
-                + " \"modelNumber\": \"7BEA\",\n"
-                + " \"lastServiceDate\": \"1535014037024\","
-                + " \"unitFamily\": \"BBU\","
-                + " \"vendorName\": \"Nokia\","
-                + " \"oamV4IpAddress\": \"10.16.123.234\","
-                + " \"softwareVersion\": \"v4.5.0.1\","
-                + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
-                + "}}}";
+            + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
+            + "\"pnfRegistrationFields\": {"
+            + " \"unitType\": \"AirScale\","
+            + " \"serialNumber\": \"QTFCOC540002E\","
+            + " \"pnfRegistrationFieldsVersion\": \"2.0\","
+            + " \"manufactureDate\": \"1535014037024\","
+            + " \"modelNumber\": \"7BEA\",\n"
+            + " \"lastServiceDate\": \"1535014037024\","
+            + " \"unitFamily\": \"BBU\","
+            + " \"vendorName\": \"Nokia\","
+            + " \"oamV4IpAddress\": \"10.16.123.234\","
+            + " \"softwareVersion\": \"v4.5.0.1\","
+            + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
+            + "}}}";
     }
 
     @Test
@@ -120,11 +121,11 @@ class DmaapConsumerTaskImplTest {
         //given
         prepareMocksForDmaapConsumer(Optional.of(message));
         //when
-        Mono<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
+        Flux<ConsumerDmaapModel> response = dmaapConsumerTask.execute("Sample input");
 
         //then
         verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
-        assertEquals(consumerDmaapModel, response.block());
+        assertEquals(consumerDmaapModel, response.blockFirst());
 
 
     }