Simplify DmaaP Publisher and consumer interfaces

Change-Id: Iaa7d870e8bd33687047832960c86adb7e97d969c
Issue-ID: DCAEGEN2-922
Signed-off-by: piotr.karas <piotr.karas@nokia.com>

8 files changed, 59 insertions, 105 deletions

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 4cde2257..2f947d47 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
@@ -20,30 +20,21 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
 import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
-import org.springframework.web.reactive.function.client.WebClient;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
  */
-abstract class DmaapConsumerTask {
+interface DmaapConsumerTask {
 
-    abstract Flux<ConsumerDmaapModel> consume(Mono<String> message);
+    void initConfigs();
 
-    abstract DMaaPConsumerReactiveHttpClient resolveClient();
+    Flux<ConsumerDmaapModel> execute(String object);
 
-    abstract void initConfigs();
+    Flux<ConsumerDmaapModel> consume(Mono<String> message);
 
-    protected abstract DmaapConsumerConfiguration resolveConfiguration();
-
-    protected abstract Flux<ConsumerDmaapModel> execute(String object);
-
-    WebClient buildWebClient() {
-        return new DMaaPReactiveWebClient().build();
-    }
+    DMaaPConsumerReactiveHttpClient resolveClient();
 }
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 3a5f213c..c4d9c44a 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
@@ -20,10 +20,9 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.prh.configuration.AppConfig;
 import org.onap.dcaegen2.services.prh.configuration.Config;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
 import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
 import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
 import org.slf4j.Logger;
@@ -37,26 +36,28 @@ import reactor.core.publisher.Mono;
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  */
 @Component
-public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
+public class DmaapConsumerTaskImpl implements DmaapConsumerTask {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DmaapConsumerTaskImpl.class);
     private final Config config;
-    private DmaapConsumerJsonParser dmaapConsumerJsonParser;
+    private final DmaapConsumerJsonParser dmaapConsumerJsonParser;
+    private final DMaaPReactiveWebClient dmaapReactiveWebClient;
 
     @Autowired
     public DmaapConsumerTaskImpl(Config config) {
-        this.config = config;
-        this.dmaapConsumerJsonParser = new DmaapConsumerJsonParser();
+        this(config, new DmaapConsumerJsonParser(), new DMaaPReactiveWebClient());
     }
 
-    DmaapConsumerTaskImpl(AppConfig prhAppConfig, DmaapConsumerJsonParser dmaapConsumerJsonParser) {
+    DmaapConsumerTaskImpl(Config prhAppConfig, DmaapConsumerJsonParser dmaapConsumerJsonParser,
+                          DMaaPReactiveWebClient dmaapReactiveWebClient) {
         this.config = prhAppConfig;
         this.dmaapConsumerJsonParser = dmaapConsumerJsonParser;
+        this.dmaapReactiveWebClient = dmaapReactiveWebClient;
     }
 
     @Override
-    Flux<ConsumerDmaapModel> consume(Mono<String> message) {
-        return dmaapConsumerJsonParser.getJsonObject(message);
+    public void initConfigs() {
+        config.initFileStreamReader();
     }
 
     @Override
@@ -67,17 +68,13 @@ public class DmaapConsumerTaskImpl extends DmaapConsumerTask {
     }
 
     @Override
-    void initConfigs() {
-        config.initFileStreamReader();
-    }
-
-    @Override
-    protected DmaapConsumerConfiguration resolveConfiguration() {
-        return config.getDmaapConsumerConfiguration();
+    public Flux<ConsumerDmaapModel> consume(Mono<String> message) {
+        return dmaapConsumerJsonParser.getJsonObject(message);
     }
 
     @Override
-    DMaaPConsumerReactiveHttpClient resolveClient() {
-        return new DMaaPConsumerReactiveHttpClient(resolveConfiguration()).createDMaaPWebClient(buildWebClient());
+    public DMaaPConsumerReactiveHttpClient resolveClient() {
+        return new DMaaPConsumerReactiveHttpClient(
+                config.getDmaapConsumerConfiguration()).createDMaaPWebClient(dmaapReactiveWebClient.build());
     }
 }
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 7a121d5f..dc0a4488 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
@@ -20,27 +20,20 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPPublisherReactiveHttpClient;
 import org.springframework.http.ResponseEntity;
-import org.springframework.web.client.RestTemplate;
 import reactor.core.publisher.Mono;
 
 /**
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 3/23/18
  */
-abstract class DmaapPublisherTask {
+interface DmaapPublisherTask {
 
-    abstract Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
+    Mono<ResponseEntity<String>> execute(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
 
-    abstract DMaaPProducerReactiveHttpClient resolveClient();
+    Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) throws PrhTaskException;
 
-    protected abstract DmaapPublisherConfiguration resolveConfiguration();
-
-    protected abstract Mono<ResponseEntity<String>> execute(ConsumerDmaapModel consumerDmaapModel)
-        throws PrhTaskException;
-
-    abstract RestTemplate buildWebClient();
+    DMaaPPublisherReactiveHttpClient resolveClient();
 }
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 9a4ff2e3..fdc5e625 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
@@ -20,11 +20,10 @@
 
 package org.onap.dcaegen2.services.prh.tasks;
 
-import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.services.prh.configuration.Config;
 import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPPublisherReactiveHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -37,11 +36,11 @@ import reactor.core.publisher.Mono;
  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18
  */
 @Component
-public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
+public class DmaapPublisherTaskImpl implements DmaapPublisherTask {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class);
     private final Config config;
-    private DMaaPProducerReactiveHttpClient dmaapProducerReactiveHttpClient;
+    private DMaaPPublisherReactiveHttpClient dmaapPublisherReactiveHttpClient;
 
     @Autowired
     public DmaapPublisherTaskImpl(Config config) {
@@ -49,32 +48,23 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask {
     }
 
     @Override
-    Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) {
-        return dmaapProducerReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel);
-    }
-
-    @Override
     public Mono<ResponseEntity<String>> execute(ConsumerDmaapModel consumerDmaapModel) throws DmaapNotFoundException {
         if (consumerDmaapModel == null) {
             throw new DmaapNotFoundException("Invoked null object to DMaaP task");
         }
-        dmaapProducerReactiveHttpClient = resolveClient();
+        dmaapPublisherReactiveHttpClient = resolveClient();
         LOGGER.info("Method called with arg {}", consumerDmaapModel);
         return publish(consumerDmaapModel);
     }
 
     @Override
-    RestTemplate buildWebClient() {
-        return new RestTemplate();
-    }
-
-    @Override
-    protected DmaapPublisherConfiguration resolveConfiguration() {
-        return config.getDmaapPublisherConfiguration();
+    public Mono<ResponseEntity<String>> publish(ConsumerDmaapModel consumerDmaapModel) {
+        return dmaapPublisherReactiveHttpClient.getDMaaPProducerResponse(consumerDmaapModel);
     }
 
     @Override
-    DMaaPProducerReactiveHttpClient resolveClient() {
-        return new DMaaPProducerReactiveHttpClient(resolveConfiguration()).createDMaaPWebClient(buildWebClient());
+    public DMaaPPublisherReactiveHttpClient resolveClient() {
+        return new DMaaPPublisherReactiveHttpClient(config.getDmaapPublisherConfiguration())
+                .createDMaaPWebClient(new RestTemplate());
     }
 }
\ No newline at end of file
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 8d6218ee..fe9f6f78 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
@@ -29,16 +29,14 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.onap.dcaegen2.services.prh.TestAppConfiguration.createDefaultDmaapConsumerConfiguration;
 
-import com.google.gson.JsonElement;
-import com.google.gson.JsonParser;
 import java.util.Optional;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 import org.onap.dcaegen2.services.prh.config.DmaapConsumerConfiguration;
 import org.onap.dcaegen2.services.prh.configuration.AppConfig;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
+import org.onap.dcaegen2.services.prh.service.DMaaPReactiveWebClient;
 import org.onap.dcaegen2.services.prh.service.DmaapConsumerJsonParser;
 import org.onap.dcaegen2.services.prh.service.consumer.DMaaPConsumerReactiveHttpClient;
 import reactor.core.publisher.Flux;
@@ -56,7 +54,6 @@ class DmaapConsumerTaskImplTest {
     private static AppConfig appConfig;
     private static DmaapConsumerConfiguration dmaapConsumerConfiguration;
     private static String message;
-    private static String parsed;
 
     @BeforeAll
     static void setUp() {
@@ -82,22 +79,6 @@ class DmaapConsumerTaskImplTest {
             + " \"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\""
-            + "}}}";
     }
 
     @Test
@@ -126,16 +107,22 @@ class DmaapConsumerTaskImplTest {
         assertEquals(consumerDmaapModel, response.blockFirst());
     }
 
+    @Test
+    void whenInitConfigs_initStreamReader() {
+        //when
+        dmaapConsumerTask.initConfigs();
+
+        //then
+        verify(appConfig).initFileStreamReader();
+    }
+
     private void prepareMocksForDmaapConsumer(Optional<String> message) {
-        DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
-        JsonElement jsonElement = new JsonParser().parse(parsed);
-        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
-            .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
         dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
         when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(message.orElse("")));
         when(appConfig.getDmaapConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
-        dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig, dmaapConsumerJsonParser));
-        when(dmaapConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration);
+        DMaaPReactiveWebClient dmaapReactiveWebClient = mock(DMaaPReactiveWebClient.class);
+        dmaapConsumerTask =
+                spy(new DmaapConsumerTaskImpl(appConfig, new DmaapConsumerJsonParser(), dmaapReactiveWebClient));
         doReturn(dMaaPConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
     }
 }
\ No newline at end of file
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
index 5f73dbe1..f42c3962 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapConsumerTaskSpy.java
@@ -50,7 +50,6 @@ public class DmaapConsumerTaskSpy {
         DmaapConsumerTaskImpl dmaapConsumerTask = spy(new DmaapConsumerTaskImpl(appConfig));
         DMaaPConsumerReactiveHttpClient dmaapConsumerReactiveHttpClient = mock(
             DMaaPConsumerReactiveHttpClient.class);
-        doReturn(mock(DmaapConsumerConfiguration.class)).when(dmaapConsumerTask).resolveConfiguration();
         doReturn(dmaapConsumerReactiveHttpClient).when(dmaapConsumerTask).resolveClient();
         return dmaapConsumerTask;
     }
diff --git a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
index 01851dc1..7ed6c5ad 100644
--- a/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
+++ b/prh-app-server/src/test/java/org/onap/dcaegen2/services/prh/tasks/DmaapProducerTaskSpy.java
@@ -26,7 +26,7 @@ import static org.mockito.Mockito.spy;
 
 import org.onap.dcaegen2.services.prh.config.DmaapPublisherConfiguration;
 import org.onap.dcaegen2.services.prh.configuration.AppConfig;
-import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPPublisherReactiveHttpClient;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Primary;
@@ -48,9 +48,8 @@ public class DmaapProducerTaskSpy {
         AppConfig appConfig = spy(AppConfig.class);
         doReturn(mock(DmaapPublisherConfiguration.class)).when(appConfig).getDmaapPublisherConfiguration();
         DmaapPublisherTaskImpl dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
-        DMaaPProducerReactiveHttpClient extendedDmaapProducerHttpClient = mock(
-            DMaaPProducerReactiveHttpClient.class);
-        doReturn(mock(DmaapPublisherConfiguration.class)).when(dmaapPublisherTask).resolveConfiguration();
+        DMaaPPublisherReactiveHttpClient extendedDmaapProducerHttpClient = mock(
+            DMaaPPublisherReactiveHttpClient.class);
         doReturn(extendedDmaapProducerHttpClient).when(dmaapPublisherTask).resolveClient();
         return dmaapPublisherTask;
     }
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 00d6fc46..e9eed698 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
@@ -40,7 +40,7 @@ import org.onap.dcaegen2.services.prh.exceptions.DmaapNotFoundException;
 import org.onap.dcaegen2.services.prh.exceptions.PrhTaskException;
 import org.onap.dcaegen2.services.prh.model.ConsumerDmaapModel;
 import org.onap.dcaegen2.services.prh.model.ImmutableConsumerDmaapModel;
-import org.onap.dcaegen2.services.prh.service.producer.DMaaPProducerReactiveHttpClient;
+import org.onap.dcaegen2.services.prh.service.producer.DMaaPPublisherReactiveHttpClient;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import reactor.core.publisher.Mono;
@@ -53,7 +53,7 @@ class DmaapPublisherTaskImplTest {
 
     private static ConsumerDmaapModel consumerDmaapModel;
     private static DmaapPublisherTaskImpl dmaapPublisherTask;
-    private static DMaaPProducerReactiveHttpClient dMaaPProducerReactiveHttpClient;
+    private static DMaaPPublisherReactiveHttpClient dMaaPPublisherReactiveHttpClient;
     private static AppConfig appConfig;
     private static DmaapPublisherConfiguration dmaapPublisherConfiguration;
 
@@ -90,9 +90,9 @@ class DmaapPublisherTaskImplTest {
             .expectNext(responseEntity).verifyComplete();
 
         //then
-        verify(dMaaPProducerReactiveHttpClient, times(1))
+        verify(dMaaPPublisherReactiveHttpClient, times(1))
             .getDMaaPProducerResponse(consumerDmaapModel);
-        verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
+        verifyNoMoreInteractions(dMaaPPublisherReactiveHttpClient);
     }
 
 
@@ -107,22 +107,20 @@ class DmaapPublisherTaskImplTest {
             .expectNext(responseEntity).verifyComplete();
 
         //then
-        verify(dMaaPProducerReactiveHttpClient, times(1))
+        verify(dMaaPPublisherReactiveHttpClient, times(1))
             .getDMaaPProducerResponse(consumerDmaapModel);
-        verifyNoMoreInteractions(dMaaPProducerReactiveHttpClient);
+        verifyNoMoreInteractions(dMaaPPublisherReactiveHttpClient);
     }
 
 
     private ResponseEntity<String> prepareMocksForTests(Integer httpResponseCode) {
         ResponseEntity<String> responseEntity = mock(ResponseEntity.class);
-        //when
         when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(httpResponseCode));
-        dMaaPProducerReactiveHttpClient = mock(DMaaPProducerReactiveHttpClient.class);
-        when(dMaaPProducerReactiveHttpClient.getDMaaPProducerResponse(any()))
+        dMaaPPublisherReactiveHttpClient = mock(DMaaPPublisherReactiveHttpClient.class);
+        when(dMaaPPublisherReactiveHttpClient.getDMaaPProducerResponse(any()))
             .thenReturn(Mono.just(responseEntity));
         dmaapPublisherTask = spy(new DmaapPublisherTaskImpl(appConfig));
-        when(dmaapPublisherTask.resolveConfiguration()).thenReturn(dmaapPublisherConfiguration);
-        doReturn(dMaaPProducerReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
+        doReturn(dMaaPPublisherReactiveHttpClient).when(dmaapPublisherTask).resolveClient();
         return responseEntity;
     }
 }
\ No newline at end of file