Bug fixes

Improve UnitTest
Add changes after review in
DmaaPReactiveTasks workflow

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

16 files changed, 83 insertions, 200 deletions

diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java
index 1d215c62..7516853e 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/service/DmaapConsumerJsonParser.java
@@ -48,28 +48,30 @@ public class DmaapConsumerJsonParser {
 
 
     public Mono<ConsumerDmaapModel> getJsonObject(Mono<String> monoMessage) {
-        return monoMessage.flatMap(message ->
-        {
-            if (!StringUtils.isEmpty(message)) {
-                JsonElement jsonElement = new JsonParser().parse(message);
-                ConsumerDmaapModel consumerDmaapModel;
-                try {
-                    if (jsonElement.isJsonObject()) {
-                        consumerDmaapModel = create(jsonElement.getAsJsonObject());
-                    } else {
-                        consumerDmaapModel = create(
-                            StreamSupport.stream(jsonElement.getAsJsonArray().spliterator(), false).findFirst()
-                                .flatMap(this::getJsonObjectFromAnArray)
-                                .orElseThrow(DmaapEmptyResponseException::new));
-                    }
-                    logger.info("Parsed model from DmaaP after getting it: {}", consumerDmaapModel);
-                    return Mono.just(consumerDmaapModel);
-                } catch (DmaapNotFoundException | DmaapEmptyResponseException e) {
-                    return Mono.error(e);
-                }
-            }
-            return Mono.error(new DmaapEmptyResponseException());
-        });
+        return monoMessage
+            .flatMap(message -> (StringUtils.isEmpty(message) ? Mono.error(new DmaapEmptyResponseException())
+                : convertJsonToConsumerDmaapModel(message)));
+    }
+
+    private Mono<? extends ConsumerDmaapModel> convertJsonToConsumerDmaapModel(String message) {
+        try {
+            JsonElement jsonElement = new JsonParser().parse(message);
+            ConsumerDmaapModel consumerDmaapModel = jsonElement.isJsonObject() ?
+                create(jsonElement.getAsJsonObject()) :
+                getConsumerDmaapModelFromJsonArray(jsonElement);
+            logger.info("Parsed model from DmaaP after getting it: {}", consumerDmaapModel);
+            return Mono.just(consumerDmaapModel);
+        } catch (DmaapNotFoundException | DmaapEmptyResponseException e) {
+            return Mono.error(e);
+        }
+    }
+
+    private ConsumerDmaapModel getConsumerDmaapModelFromJsonArray(JsonElement jsonElement)
+        throws DmaapNotFoundException, DmaapEmptyResponseException {
+        return create(
+            StreamSupport.stream(jsonElement.getAsJsonArray().spliterator(), false).findFirst()
+                .flatMap(this::getJsonObjectFromAnArray)
+                .orElseThrow(DmaapEmptyResponseException::new));
     }
 
     public Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) {
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
index 005d08d1..b12fb5bb 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/AAIProducerTaskImpl.java
@@ -26,8 +26,8 @@ import org.onap.dcaegen2.services.prh.configuration.AppConfig;
 import org.onap.dcaegen2.services.prh.configuration.Config;
 import org.onap.dcaegen2.services.prh.exceptions.AAINotFoundException;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.model.utils.HttpUtils;
 import org.onap.dcaegen2.services.prh.service.AAIProducerClient;
-import org.onap.dcaegen2.services.prh.service.HttpUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
index 8df564d2..93c287b4 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTask.java
@@ -43,10 +43,6 @@ abstract class DmaapConsumerTask {
     protected abstract Mono<ConsumerDmaapModel> execute(String object) throws PrhTaskException;
 
     WebClient buildWebClient() {
-        DmaapConsumerConfiguration dmaapConsumerConfiguration = resolveConfiguration();
-        return new DMaaPReactiveWebClient.WebClientBuilder()
-            .dmaapContentType(dmaapConsumerConfiguration.dmaapContentType())
-            .dmaapUserName(dmaapConsumerConfiguration.dmaapUserName())
-            .dmaapUserPassword(dmaapConsumerConfiguration.dmaapUserPassword()).build();
+        return new DMaaPReactiveWebClient().fromConfiguration(resolveConfiguration()).build();
     }
 }
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
index 8c74bac3..45709aa2 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskImpl.java
@@ -19,7 +19,6 @@
  */
 package org.onap.dcaegen2.services.prh.tasks;
 
-import java.util.Optional;
 import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
 import org.onap.dcaegen2.services.prh.configuration.AppConfig;
 import org.onap.dcaegen2.services.prh.configuration.Config;
@@ -79,8 +78,8 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
 
     @Override
     DMaaPConsumerReactiveHttpClient resolveClient() {
-        return Optional.ofNullable(dMaaPConsumerReactiveHttpClient)
-            .orElseGet(() -> new DMaaPConsumerReactiveHttpClient(resolveConfiguration()).createDMaaPWebClient(
-                buildWebClient()));
+        return dMaaPConsumerReactiveHttpClient == null
+            ? new DMaaPConsumerReactiveHttpClient(resolveConfiguration()).createDMaaPWebClient(buildWebClient())
+            : dMaaPConsumerReactiveHttpClient;
     }
 }
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
index af8d14a3..f559683d 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTask.java
@@ -32,19 +32,15 @@ import reactor.core.publisher.Mono;
  */
 abstract class DmaapPublisherTask {
 
-    abstract Mono<Integer> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
+    abstract Mono<String> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
 
     abstract DMaaPProducerReactiveHttpClient resolveClient();
 
     protected abstract DmaapPublisherConfiguration resolveConfiguration();
 
-    protected abstract Mono<Integer> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
+    protected abstract Mono<String> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws PrhTaskException;
 
     WebClient buildWebClient() {
-        DmaapPublisherConfiguration dmaapPublisherConfiguration = resolveConfiguration();
-        return new DMaaPReactiveWebClient.WebClientBuilder()
-            .dmaapContentType(dmaapPublisherConfiguration.dmaapContentType())
-            .dmaapUserName(dmaapPublisherConfiguration.dmaapUserName())
-            .dmaapUserPassword(dmaapPublisherConfiguration.dmaapUserPassword()).build();
+        return new DMaaPReactiveWebClient().fromConfiguration(resolveConfiguration()).build();
     }
 }
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
index 11281d81..673e00f3 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImpl.java
@@ -48,14 +48,14 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
     }
 
     @Override
-    Mono<Integer> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) {
+    Mono<String> publish(Mono<ConsumerDmaapModel> consumerDmaapModel) {
         logger.info("Publishing on DMaaP topic {} object {}", resolveConfiguration().dmaapTopicName(),
             consumerDmaapModel);
-        return dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel).map(Integer::parseInt);
+        return dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel);
     }
 
     @Override
-    public Mono<Integer> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws DmaapNotFoundException {
+    public Mono<String> execute(Mono<ConsumerDmaapModel> consumerDmaapModel) throws DmaapNotFoundException {
         consumerDmaapModel = Optional.ofNullable(consumerDmaapModel)
             .orElseThrow(() -> new DmaapNotFoundException("Invoked null object to DMaaP task"));
         dMaaPProducerReactiveHttpClient = resolveClient();
@@ -70,8 +70,8 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
 
     @Override
     DMaaPProducerReactiveHttpClient resolveClient() {
-        return Optional.ofNullable(dMaaPProducerReactiveHttpClient)
-            .orElseGet(() -> new DMaaPProducerReactiveHttpClient(resolveConfiguration())
-                .createDMaaPWebClient(buildWebClient()));
+        return dMaaPProducerReactiveHttpClient == null
+            ? new DMaaPProducerReactiveHttpClient(resolveConfiguration()).createDMaaPWebClient(buildWebClient())
+            : dMaaPProducerReactiveHttpClient;
     }
 }
\ No newline at end of file
diff --git a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
index 2787e64b..365552b7 100644
--- a/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
+++ b/prh-app-server/src/main/java/org/onap/dcaegen2/services/prh/tasks/ScheduledTasks.java
@@ -54,7 +54,7 @@ public class ScheduledTasks {
     public void scheduleMainPrhEventTask() {
         logger.trace("Execution of tasks was registered");
 
-        Mono<Integer> dmaapProducerResponse = Mono.fromCallable(consumeFromDMaaPMessage())
+        Mono<String> dmaapProducerResponse = Mono.fromCallable(consumeFromDMaaPMessage())
             .doOnError(DmaapEmptyResponseException.class, error -> logger.warn("Nothing to consume from DMaaP"))
             .map(this::publishToAAIConfiguration)
             .flatMap(this::publishToDMaaPConfiguration)
@@ -67,7 +67,7 @@ public class ScheduledTasks {
         logger.info("PRH tasks have been completed");
     }
 
-    private void onSuccess(Integer responseCode) {
+    private void onSuccess(String responseCode) {
         logger.info("Prh consumed tasks. HTTP Response code {}", responseCode);
     }
 
@@ -90,17 +90,15 @@ public class ScheduledTasks {
             try {
                 return Mono.just(aaiProducerTask.execute(dmaapModel));
             } catch (PrhTaskException e) {
-                logger.warn("Exception in A&AIProducer task ", e);
                 return Mono.error(e);
             }
         });
     }
 
-    private Mono<Integer> publishToDMaaPConfiguration(Mono<ConsumerDmaapModel> monoAAIModel) {
+    private Mono<String> publishToDMaaPConfiguration(Mono<ConsumerDmaapModel> monoAAIModel) {
         try {
             return dmaapProducerTask.execute(monoAAIModel);
         } catch (PrhTaskException e) {
-            logger.warn("Exception in DMaaPProducer task ", e);
             return Mono.error(e);
         }
     }
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 9e7edc4d..32f5c1e2 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
@@ -180,7 +180,7 @@ class DmaapConsumerJsonParserTest {
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
-            .expectSubscription().expectError(DmaapNotFoundException.class);
+            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
 
     }
 
@@ -202,7 +202,7 @@ class DmaapConsumerJsonParserTest {
                 + ":\"Normal\",\"reportingEntityName\":\"5GRAN_DU\",\"sequence\":0,\"sourceId\":\"<<SerialNumber>>\","
                 + "\"sourceName\":\"5GRAN_DU\",\"startEpochMicrosec\":1519837825682,\"version\":3}}}]";
         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessage)))
-            .expectSubscription().expectError(DmaapNotFoundException.class);
+            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
@@ -232,7 +232,7 @@ class DmaapConsumerJsonParserTest {
                 + "\"pnfType\":\"AirScale\"}}}]";
         StepVerifier
             .create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutPnfVendorAndSerialNumber)))
-            .expectSubscription().expectError(DmaapNotFoundException.class);
+            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
     }
 
     @Test
@@ -260,6 +260,6 @@ class DmaapConsumerJsonParserTest {
                 + "\"AJ02\",\"pnfSerialNumber\":\"QTFCOC540002E\",\"pnfSoftwareVersion\":\"v4.5.0.1\",\"pnfType\":\"AirScale\","
                 + "\"pnfVendorName\":\"Nokia\"}}}]";
         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIPInformation)))
-            .expectSubscription().expectError(DmaapNotFoundException.class);
+            .expectSubscription().expectError(DmaapNotFoundException.class).verify();
     }
 }
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 c5a9d85a..b903fc90 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
@@ -19,6 +19,7 @@
  */
 package org.onap.dcaegen2.services.prh.tasks;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -98,7 +99,7 @@ class DmaapConsumerTaskImplTest {
 
         //then
         StepVerifier.create(dmaapConsumerTask.execute("Sample input")).expectSubscription()
-            .expectError(DmaapEmptyResponseException.class);
+            .expectError(DmaapEmptyResponseException.class).verify();
 
         verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
     }
@@ -112,7 +113,7 @@ class DmaapConsumerTaskImplTest {
 
         //then
         verify(dMaaPConsumerReactiveHttpClient, times(1)).getDMaaPConsumerResponse();
-        Assertions.assertEquals(consumerDmaapModel, response.block());
+        assertEquals(consumerDmaapModel, response.block());
 
 
     }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
index 6b08be55..25d1fff6 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapPublisherTaskImplTest.java
@@ -19,6 +19,7 @@
  */
 package org.onap.dcaegen2.services.prh.tasks;
 
+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;
@@ -28,7 +29,6 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.function.Executable;
@@ -77,8 +77,7 @@ class DmaapPublisherTaskImplTest {
         Executable executableFunction = () -> dmaapPublisherTask.execute(null);
 
         //then
-        Assertions
-            .assertThrows(PrhTaskException.class, executableFunction, "The specified parameter is incorrect");
+        assertThrows(PrhTaskException.class, executableFunction, "The specified parameter is incorrect");
     }
 
     @Test
@@ -88,7 +87,7 @@ class DmaapPublisherTaskImplTest {
 
         //when
         StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
-            .expectNext(HttpStatus.OK.value());
+            .expectNext(HttpStatus.OK.toString()).verifyComplete();
 
         //then
         verify(dMaaPProducerReactiveHttpClient, times(1))
@@ -96,6 +95,7 @@ class DmaapPublisherTaskImplTest {
         verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
     }
 
+
     @Test
     public void whenPassedObjectFits_butIncorrectResponseReturns() throws DmaapNotFoundException {
         //given
@@ -103,7 +103,7 @@ class DmaapPublisherTaskImplTest {
 
         //when
         StepVerifier.create(dmaapPublisherTask.execute(Mono.just(consumerDmaapModel))).expectSubscription()
-            .expectError(PrhTaskException.class);
+            .expectNext(String.valueOf(HttpStatus.UNAUTHORIZED.value())).verifyComplete();
 
         //then
         verify(dMaaPProducerReactiveHttpClient, times(1)).getDMaaPProducerResponse(any(Mono.class));
@@ -114,7 +114,7 @@ class DmaapPublisherTaskImplTest {
     private void prepareMocksForTests(Integer httpResponseCode) {
         dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class);
         when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any(Mono.class)))
-            .thenReturn(Mono.just(httpResponseCode));
+            .thenReturn(Mono.just(httpResponseCode.toString()));
         dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
         when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration);
         doReturn(dMaaPProducerReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
diff --git a/prh-dmaap-client/pom.xml b/prh-dmaap-client/pom.xml
index 0633b46a..037bfe88 100644
--- a/prh-dmaap-client/pom.xml
+++ b/prh-dmaap-client/pom.xml
@@ -59,10 +59,6 @@
       <version>2.0.4.RELEASE</version>
     </dependency>
     <dependency>
-      <groupId>org.apache.httpcomponents</groupId>
-      <artifactId>httpclient</artifactId>
-    </dependency>
-    <dependency>
       <groupId>org.onap.dcaegen2.services.prh</groupId>
       <artifactId>prh-commons</artifactId>
       <version>1.0.0-SNAPSHOT</version>
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java
index a41ec3a4..ab81bede 100644
--- a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java
+++ b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClient.java
@@ -21,6 +21,7 @@ package org.onap.dcaegen2.services.prh.service;
 
 import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
 
+import org.onap.dcaegen2.services.prh.config.DmaapCustomConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.http.HttpHeaders;
@@ -35,26 +36,34 @@ public class DMaaPReactiveWebClient {
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
-    private DMaaPReactiveWebClient() {
+    private String dMaaPContentType;
+    private String dMaaPUserName;
+    private String dMaaPUserPassword;
+
+    public DMaaPReactiveWebClient fromConfiguration(DmaapCustomConfig dmaapCustomConfig) {
+        this.dMaaPUserName = dmaapCustomConfig.dmaapUserName();
+        this.dMaaPUserPassword = dmaapCustomConfig.dmaapUserPassword();
+        this.dMaaPContentType = dmaapCustomConfig.dmaapContentType();
+        return this;
     }
 
-    private WebClient create(WebClientBuilder webClientBuilder) {
+    public WebClient build() {
         return WebClient.builder()
-            .defaultHeader(HttpHeaders.CONTENT_TYPE, webClientBuilder.dMaaPContentType)
-            .filter(basicAuthentication(webClientBuilder.dMaaPUserName, webClientBuilder.dMaaPUserPassword))
+            .defaultHeader(HttpHeaders.CONTENT_TYPE, dMaaPContentType)
+            .filter(basicAuthentication(dMaaPUserName, dMaaPUserPassword))
             .filter(logRequest())
             .filter(logResponse())
             .build();
     }
 
-    ExchangeFilterFunction logResponse() {
+    private ExchangeFilterFunction logResponse() {
         return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
             logger.info("Response Status {}", clientResponse.statusCode());
             return Mono.just(clientResponse);
         });
     }
 
-    ExchangeFilterFunction logRequest() {
+    private ExchangeFilterFunction logRequest() {
         return ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
             logger.info("Request: {} {}", clientRequest.method(), clientRequest.url());
             clientRequest.headers()
@@ -63,32 +72,4 @@ public class DMaaPReactiveWebClient {
         });
     }
 
-    public static class WebClientBuilder {
-
-        private String dMaaPContentType;
-        private String dMaaPUserName;
-        private String dMaaPUserPassword;
-
-        public WebClientBuilder() {
-        }
-
-        public WebClientBuilder dmaapContentType(String dmaapContentType) {
-            this.dMaaPContentType = dmaapContentType;
-            return this;
-        }
-
-        public WebClientBuilder dmaapUserName(String dmaapUserName) {
-            this.dMaaPUserName = dmaapUserName;
-            return this;
-        }
-
-        public WebClientBuilder dmaapUserPassword(String dmaapUserPassword) {
-            this.dMaaPUserPassword = dmaapUserPassword;
-            return this;
-        }
-
-        public WebClient build() {
-            return new DMaaPReactiveWebClient().create(this);
-        }
-    }
 }
diff --git a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/HttpUtils.java b/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/HttpUtils.java
deleted file mode 100644
index b96f80dd..00000000
--- a/prh-dmaap-client/src/main/java/org/onap/dcaegen2/services/prh/service/HttpUtils.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- * PNF-REGISTRATION-HANDLER
- * ================================================================================
- * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
- * ================================================================================
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- * ============LICENSE_END=========================================================
- */
-package org.onap.dcaegen2.services.prh.service;
-
-import org.apache.http.HttpStatus;
-
-public final class HttpUtils implements HttpStatus {
-
-    private HttpUtils() {}
-
-    public static boolean isSuccessfulResponseCode(Integer statusCode) {
-        return statusCode >= 200 && statusCode < 300;
-    }
-}
diff --git a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClientTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClientTest.java
index fad18699..2e323c5e 100644
--- a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClientTest.java
+++ b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/DMaaPReactiveWebClientTest.java
@@ -19,8 +19,12 @@
  */
 package org.onap.dcaegen2.services.prh.service;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
 import org.springframework.web.reactive.function.client.WebClient;
 
 /**
@@ -28,19 +32,23 @@ import org.springframework.web.reactive.function.client.WebClient;
  */
 public class DMaaPReactiveWebClientTest {
 
+
     @Test
     public void builder_shouldBuildDMaaPReactiveWebClient() {
         //given
+        DmaapConsumerConfiguration dmaapConsumerConfiguration = mock(DmaapConsumerConfiguration.class);
         WebClient dMaaPReactiveWebClient;
         String dMaaPContentType = "*/*";
         String dMaaPUserName = "DMaaP";
         String dMaaPUserPassword = "DMaaP";
 
         //when
-        dMaaPReactiveWebClient = new DMaaPReactiveWebClient.WebClientBuilder()
-            .dmaapContentType(dMaaPContentType)
-            .dmaapUserName(dMaaPUserName)
-            .dmaapUserPassword(dMaaPUserPassword).build();
+        when(dmaapConsumerConfiguration.dmaapContentType()).thenReturn(dMaaPContentType);
+        when(dmaapConsumerConfiguration.dmaapUserName()).thenReturn(dMaaPUserName);
+        when(dmaapConsumerConfiguration.dmaapUserPassword()).thenReturn(dMaaPUserPassword);
+        dMaaPReactiveWebClient = new DMaaPReactiveWebClient()
+            .fromConfiguration(dmaapConsumerConfiguration)
+            .build();
 
         //then
         Assertions.assertNotNull(dMaaPReactiveWebClient);
diff --git a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
index cbc7bd62..1f04da15 100644
--- a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
+++ b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/consumer/DMaaPConsumerReactiveHttpClientTest.java
@@ -98,36 +98,6 @@ public class DMaaPConsumerReactiveHttpClientTest {
             }).verifyComplete();
     }
 
-
-    @Test
-    public void getHttpResponse_HttpResponse4xxClientError() {
-
-        //when
-        mockDependantObjects();
-        doAnswer(invocationOnMock -> Mono.error(new Exception("400")))
-            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("400")));
-        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
-
-        //then
-        StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
-            .expectError(Exception.class);
-
-    }
-
-    @Test
-    public void getHttpResponse_HttpResponse5xxClientError() {
-
-        //when
-        mockDependantObjects();
-        doAnswer(invocationOnMock -> Mono.error(new Exception("500")))
-            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("500")));
-        DMaaPConsumerReactiveHttpClient.createDMaaPWebClient(webClient);
-
-        //then
-        StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
-            .expectError(Exception.class);
-    }
-
     @Test
     public void getHttpResponse_whenURISyntaxExceptionHasBeenThrown() throws URISyntaxException {
         //given
@@ -139,7 +109,7 @@ public class DMaaPConsumerReactiveHttpClientTest {
 
         //then
         StepVerifier.create(DMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).expectSubscription()
-            .expectError(Exception.class);
+            .expectError(Exception.class).verify();
     }
 
     private void mockDependantObjects() {
diff --git a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java
index f1fb70b0..20ab64e5 100644
--- a/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java
+++ b/prh-dmaap-client/src/test/java/org/onap/dcaegen2/services/prh/service/producer/DMaaPProducerReactiveHttpClientTest.java
@@ -20,7 +20,6 @@
 package org.onap.dcaegen2.services.prh.service.producer;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -36,7 +35,6 @@ import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModelForUnitTest;
 import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClient.RequestBodyUriSpec;
 import org.springframework.web.reactive.function.client.WebClient.RequestHeadersSpec;
@@ -97,37 +95,6 @@ public class DMaaPProducerReactiveHttpClientTest {
     }
 
     @Test
-    public void getHttpResponse_HttpResponse4xxClientError() {
-        //when
-        mockWebClientDependantObject();
-
-        doAnswer(invocationOnMock -> Mono.error(new Exception("400")))
-            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("400")));
-        dMaaPProducerReactiveHttpClient.createDMaaPWebClient(webClient);
-
-        //then
-        StepVerifier.create(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(Mono.just(consumerDmaapModel)))
-            .expectSubscription()
-            .expectError(Exception.class);
-
-    }
-
-    @Test
-    public void getHttpResponse_HttpResponse5xxClientError() {
-
-        //when
-        mockWebClientDependantObject();
-        doAnswer(invocationOnMock -> Mono.error(new Exception("500")))
-            .when(responseSpec).onStatus(HttpStatus::is4xxClientError, e -> Mono.error(new Exception("500")));
-        dMaaPProducerReactiveHttpClient.createDMaaPWebClient(webClient);
-
-        //then
-        StepVerifier.create(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(Mono.just(consumerDmaapModel)))
-            .expectSubscription()
-            .expectError(Exception.class);
-    }
-
-    @Test
     public void getHttpResponse_whenURISyntaxExceptionHasBeenThrown() throws URISyntaxException {
         //given
         dMaaPProducerReactiveHttpClient = spy(dMaaPProducerReactiveHttpClient);
@@ -138,7 +105,7 @@ public class DMaaPProducerReactiveHttpClientTest {
 
         //then
         StepVerifier.create(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any())).expectSubscription()
-            .expectError(Exception.class);
+            .expectError(Exception.class).verify();
     }
 
     private void mockWebClientDependantObject() {