Upgrade of BBS-ep service

Upgraded service to use latest DCAE-SDK
Upgraded many of the dependencies to latest versions
Introduced Java 11

Change-Id: I29d265d2a75aa80749f567cfb10920b2c45c2cec
Issue-ID: DCAEGEN2-2105
Signed-off-by: Stavros Kanarakis <stavros.kanarakis@nokia.com>

14 files changed, 727 insertions, 834 deletions

diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ApplicationConfigurationTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ApplicationConfigurationTest.java
index 69fbb3f0..2d9e49fd 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ApplicationConfigurationTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ApplicationConfigurationTest.java
@@ -37,8 +37,6 @@ import org.onap.bbs.event.processor.model.GeneratedAppConfigObject;
 import org.onap.bbs.event.processor.model.ImmutableDmaapInfo;
 import org.onap.bbs.event.processor.model.ImmutableGeneratedAppConfigObject;
 import org.onap.bbs.event.processor.model.ImmutableStreamsObject;
-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.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -66,6 +64,8 @@ import org.springframework.test.context.TestPropertySource;
         "configs.aai.client.aaiHeaders.Content-Type=application/merge-patch+json",
         "configs.dmaap.consumer.re-registration.dmaapHostName=test localhost",
         "configs.dmaap.consumer.re-registration.dmaapPortNumber=1234",
+        "configs.dmaap.consumer.re-registration.dmaapUserName=",
+        "configs.dmaap.consumer.re-registration.dmaapUserPassword=",
         "configs.dmaap.consumer.re-registration.dmaapTopicName=/events/unauthenticated.PNF_REREGISTRATION",
         "configs.dmaap.consumer.re-registration.dmaapProtocol=http",
         "configs.dmaap.consumer.re-registration.dmaapContentType=application/json",
@@ -75,6 +75,8 @@ import org.springframework.test.context.TestPropertySource;
         "configs.dmaap.consumer.re-registration.messageLimit=1",
         "configs.dmaap.consumer.cpe-authentication.dmaapHostName=test localhost",
         "configs.dmaap.consumer.cpe-authentication.dmaapPortNumber=1234",
+        "configs.dmaap.consumer.cpe-authentication.dmaapUserName=",
+        "configs.dmaap.consumer.cpe-authentication.dmaapUserPassword=",
         "configs.dmaap.consumer.cpe-authentication.dmaapTopicName=/events/unauthenticated.CPE_AUTHENTICATION",
         "configs.dmaap.consumer.cpe-authentication.dmaapProtocol=http",
         "configs.dmaap.consumer.cpe-authentication.dmaapContentType=application/json",
@@ -84,13 +86,15 @@ import org.springframework.test.context.TestPropertySource;
         "configs.dmaap.consumer.cpe-authentication.messageLimit=1",
         "configs.dmaap.producer.dmaapHostName=test localhost",
         "configs.dmaap.producer.dmaapPortNumber=1234",
+        "configs.dmaap.producer.dmaapUserName=",
+        "configs.dmaap.producer.dmaapUserPassword=",
         "configs.dmaap.producer.dmaapTopicName=/events/unauthenticated.DCAE_CL_OUTPUT",
         "configs.dmaap.producer.dmaapProtocol=http",
         "configs.dmaap.producer.dmaapContentType=application/json",
-        "configs.security.trustStorePath=test trust store path",
-        "configs.security.trustStorePasswordPath=test trust store password path",
-        "configs.security.keyStorePath=test key store path",
-        "configs.security.keyStorePasswordPath=test key store password path",
+        "configs.security.trustStorePath=KeyStore.jks",
+        "configs.security.trustStorePasswordPath=KeyStorePass.txt",
+        "configs.security.keyStorePath=KeyStore.jks",
+        "configs.security.keyStorePasswordPath=KeyStorePass.txt",
         "configs.security.enableDmaapCertAuth=false",
         "configs.security.enableAaiCertAuth=false",
         "configs.application.pipelinesPollingIntervalSec=30",
@@ -132,7 +136,7 @@ class ApplicationConfigurationTest {
     @Test
     void testA_configurationObjectSuccessfullyPopulated() {
 
-        AaiClientConfiguration aaiClientConfiguration = configuration.getAaiClientConfiguration();
+        var aaiClientConfiguration = configuration.getAaiClientConfiguration();
         assertAll("AAI Client Configuration Properties",
             () -> assertEquals("test localhost", aaiClientConfiguration.aaiHost()),
             () -> assertEquals(Integer.valueOf(1234), aaiClientConfiguration.aaiPort()),
@@ -148,50 +152,64 @@ class ApplicationConfigurationTest {
                         aaiClientConfiguration.aaiHeaders().get("Content-Type"))
         );
 
-        DmaapConsumerConfiguration dmaapConsumerReRegistrationConfig =
-                configuration.getDmaapReRegistrationConsumerConfiguration();
         assertAll("DMaaP Consumer Re-Registration Configuration Properties",
-            () -> assertEquals("test localhost", dmaapConsumerReRegistrationConfig.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(1234), dmaapConsumerReRegistrationConfig.dmaapPortNumber()),
+            () -> assertEquals("test localhost",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapHostName()),
+            () -> assertEquals(1234,
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapPortNumber()),
             () -> assertEquals("/events/unauthenticated.PNF_REREGISTRATION",
-                    dmaapConsumerReRegistrationConfig.dmaapTopicName()),
-            () -> assertEquals("http", dmaapConsumerReRegistrationConfig.dmaapProtocol()),
-            () -> assertEquals("", dmaapConsumerReRegistrationConfig.dmaapUserName()),
-            () -> assertEquals("", dmaapConsumerReRegistrationConfig.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapConsumerReRegistrationConfig.dmaapContentType()),
-            () -> assertEquals("c12", dmaapConsumerReRegistrationConfig.consumerId()),
-            () -> assertEquals("OpenDcae-c12", dmaapConsumerReRegistrationConfig.consumerGroup()),
-            () -> assertEquals(Integer.valueOf(-1), dmaapConsumerReRegistrationConfig.timeoutMs()),
-            () -> assertEquals(Integer.valueOf(1), dmaapConsumerReRegistrationConfig.messageLimit())
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapTopicName()),
+            () -> assertEquals("http",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapProtocol()),
+            () -> assertEquals("",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapUserName()),
+            () -> assertEquals("",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapContentType()),
+            () -> assertEquals("c12",
+                    configuration.getDmaapReRegistrationConsumerProperties().getConsumerId()),
+            () -> assertEquals("OpenDcae-c12",
+                    configuration.getDmaapReRegistrationConsumerProperties().getConsumerGroup()),
+            () -> assertEquals(-1, configuration.getDmaapReRegistrationConsumerProperties().getTimeoutMs()),
+            () -> assertEquals(1, configuration.getDmaapReRegistrationConsumerProperties().getMessageLimit())
         );
 
-        DmaapConsumerConfiguration dmaapConsumerCpeAuthenticationConfig =
-                configuration.getDmaapCpeAuthenticationConsumerConfiguration();
         assertAll("DMaaP Consumer CPE Authentication Configuration Properties",
-            () -> assertEquals("test localhost", dmaapConsumerCpeAuthenticationConfig.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(1234), dmaapConsumerCpeAuthenticationConfig.dmaapPortNumber()),
+            () -> assertEquals("test localhost",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapHostName()),
+            () -> assertEquals(1234,
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapPortNumber()),
             () -> assertEquals("/events/unauthenticated.CPE_AUTHENTICATION",
-                    dmaapConsumerCpeAuthenticationConfig.dmaapTopicName()),
-            () -> assertEquals("http", dmaapConsumerCpeAuthenticationConfig.dmaapProtocol()),
-            () -> assertEquals("", dmaapConsumerCpeAuthenticationConfig.dmaapUserName()),
-            () -> assertEquals("", dmaapConsumerCpeAuthenticationConfig.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapConsumerCpeAuthenticationConfig.dmaapContentType()),
-            () -> assertEquals("c12", dmaapConsumerCpeAuthenticationConfig.consumerId()),
-            () -> assertEquals("OpenDcae-c12", dmaapConsumerCpeAuthenticationConfig.consumerGroup()),
-            () -> assertEquals(Integer.valueOf(-1), dmaapConsumerCpeAuthenticationConfig.timeoutMs()),
-            () -> assertEquals(Integer.valueOf(1), dmaapConsumerCpeAuthenticationConfig.messageLimit())
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapTopicName()),
+            () -> assertEquals("http",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapProtocol()),
+            () -> assertEquals("",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapUserName()),
+            () -> assertEquals("",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapContentType()),
+            () -> assertEquals("c12",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getConsumerId()),
+            () -> assertEquals("OpenDcae-c12",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getConsumerGroup()),
+            () -> assertEquals(-1, configuration.getDmaapCpeAuthenticationConsumerProperties().getTimeoutMs()),
+            () -> assertEquals(1,
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getMessageLimit())
         );
 
-        DmaapPublisherConfiguration dmaapPublisherConfiguration = configuration.getDmaapPublisherConfiguration();
         assertAll("DMaaP Publisher Configuration Properties",
-            () -> assertEquals("test localhost", dmaapPublisherConfiguration.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(1234), dmaapPublisherConfiguration.dmaapPortNumber()),
+            () -> assertEquals("test localhost",
+                    configuration.getDmaapProducerProperties().getDmaapHostName()),
+            () -> assertEquals(1234, configuration.getDmaapProducerProperties().getDmaapPortNumber()),
             () -> assertEquals("/events/unauthenticated.DCAE_CL_OUTPUT",
-                    dmaapPublisherConfiguration.dmaapTopicName()),
-            () -> assertEquals("http", dmaapPublisherConfiguration.dmaapProtocol()),
-            () -> assertEquals("", dmaapPublisherConfiguration.dmaapUserName()),
-            () -> assertEquals("", dmaapPublisherConfiguration.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapPublisherConfiguration.dmaapContentType())
+                    configuration.getDmaapProducerProperties().getDmaapTopicName()),
+            () -> assertEquals("http", configuration.getDmaapProducerProperties().getDmaapProtocol()),
+            () -> assertEquals("", configuration.getDmaapProducerProperties().getDmaapUserName()),
+            () -> assertEquals("", configuration.getDmaapProducerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapProducerProperties().getDmaapContentType())
         );
 
         assertAll("Generic Application Properties",
@@ -211,12 +229,11 @@ class ApplicationConfigurationTest {
 
         assertAll("Security Application Properties",
             () -> assertFalse(aaiClientConfiguration.enableAaiCertAuth()),
-            () -> assertFalse(dmaapConsumerReRegistrationConfig.enableDmaapCertAuth()),
-            () -> assertEquals("test key store path", aaiClientConfiguration.keyStorePath()),
-            () -> assertEquals("test key store password path",
+            () -> assertEquals("KeyStore.jks", aaiClientConfiguration.keyStorePath()),
+            () -> assertEquals("KeyStorePass.txt",
                         aaiClientConfiguration.keyStorePasswordPath()),
-            () -> assertEquals("test trust store path", aaiClientConfiguration.trustStorePath()),
-            () -> assertEquals("test trust store password path",
+            () -> assertEquals("KeyStore.jks", aaiClientConfiguration.trustStorePath()),
+            () -> assertEquals("KeyStorePass.txt",
                         aaiClientConfiguration.trustStorePasswordPath())
         );
     }
@@ -298,10 +315,10 @@ class ApplicationConfigurationTest {
                 .cpeAuthConfigKey("config_key_2")
                 .closeLoopConfigKey("config_key_3")
                 .loggingLevel("TRACE")
-                .keyStorePath("test key store path - update")
-                .keyStorePasswordPath("test key store password path - update")
-                .trustStorePath("test trust store path - update")
-                .trustStorePasswordPath("test trust store password path - update")
+                .keyStorePath("KeyStore-update.jks")
+                .keyStorePasswordPath("KeyStorePass-update.txt")
+                .trustStorePath("KeyStore-update.jks")
+                .trustStorePasswordPath("KeyStorePass-update.txt")
                 .enableAaiCertAuth(true)
                 .enableDmaapCertAuth(true)
                 .streamSubscribesMap(subscribes)
@@ -311,7 +328,7 @@ class ApplicationConfigurationTest {
         // Update the configuration
         configuration.updateCurrentConfiguration(updatedConfiguration);
 
-        AaiClientConfiguration aaiClientConfiguration = configuration.getAaiClientConfiguration();
+        var aaiClientConfiguration = configuration.getAaiClientConfiguration();
         assertAll("AAI Client Configuration Properties",
             () -> assertEquals("aai.onap.svc.cluster.local", aaiClientConfiguration.aaiHost()),
             () -> assertEquals(Integer.valueOf(8443), aaiClientConfiguration.aaiPort()),
@@ -327,50 +344,65 @@ class ApplicationConfigurationTest {
                         aaiClientConfiguration.aaiHeaders().get("Content-Type"))
         );
 
-        DmaapConsumerConfiguration dmaapConsumerReRegistrationConfig =
-                configuration.getDmaapReRegistrationConsumerConfiguration();
         assertAll("DMaaP Consumer Re-Registration Configuration Properties",
-            () -> assertEquals("we-are-message-router1.us", dmaapConsumerReRegistrationConfig.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(3901), dmaapConsumerReRegistrationConfig.dmaapPortNumber()),
-            () -> assertEquals("events/unauthenticated.PNF_UPDATE",
-                    dmaapConsumerReRegistrationConfig.dmaapTopicName()),
-            () -> assertEquals("https", dmaapConsumerReRegistrationConfig.dmaapProtocol()),
-            () -> assertEquals("some-user", dmaapConsumerReRegistrationConfig.dmaapUserName()),
-            () -> assertEquals("some-password", dmaapConsumerReRegistrationConfig.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapConsumerReRegistrationConfig.dmaapContentType()),
-            () -> assertEquals("c13", dmaapConsumerReRegistrationConfig.consumerId()),
-            () -> assertEquals("OpenDcae-c13", dmaapConsumerReRegistrationConfig.consumerGroup()),
-            () -> assertEquals(Integer.valueOf(5), dmaapConsumerReRegistrationConfig.timeoutMs()),
-            () -> assertEquals(Integer.valueOf(10), dmaapConsumerReRegistrationConfig.messageLimit())
+            () -> assertEquals("we-are-message-router1.us",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapHostName()),
+            () -> assertEquals(3901,
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapPortNumber()),
+            () -> assertEquals("unauthenticated.PNF_UPDATE",
+                        configuration.getDmaapReRegistrationConsumerProperties().getDmaapTopicName()),
+            () -> assertEquals("https",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapProtocol()),
+            () -> assertEquals("some-user",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapUserName()),
+            () -> assertEquals("some-password",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapReRegistrationConsumerProperties().getDmaapContentType()),
+            () -> assertEquals("c13",
+                    configuration.getDmaapReRegistrationConsumerProperties().getConsumerId()),
+            () -> assertEquals("OpenDcae-c13",
+                    configuration.getDmaapReRegistrationConsumerProperties().getConsumerGroup()),
+            () -> assertEquals(5, configuration.getDmaapReRegistrationConsumerProperties().getTimeoutMs()),
+            () -> assertEquals(10, configuration.getDmaapReRegistrationConsumerProperties().getMessageLimit())
         );
 
-        DmaapConsumerConfiguration dmaapConsumerCpeAuthenticationConfig =
-                configuration.getDmaapCpeAuthenticationConsumerConfiguration();
         assertAll("DMaaP Consumer CPE Authentication Configuration Properties",
-            () -> assertEquals("we-are-message-router2.us", dmaapConsumerCpeAuthenticationConfig.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(3902), dmaapConsumerCpeAuthenticationConfig.dmaapPortNumber()),
-            () -> assertEquals("events/unauthenticated.CPE_AUTHENTICATION",
-                    dmaapConsumerCpeAuthenticationConfig.dmaapTopicName()),
-            () -> assertEquals("https", dmaapConsumerCpeAuthenticationConfig.dmaapProtocol()),
-            () -> assertEquals("some-user", dmaapConsumerCpeAuthenticationConfig.dmaapUserName()),
-            () -> assertEquals("some-password", dmaapConsumerCpeAuthenticationConfig.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapConsumerCpeAuthenticationConfig.dmaapContentType()),
-            () -> assertEquals("c13", dmaapConsumerCpeAuthenticationConfig.consumerId()),
-            () -> assertEquals("OpenDcae-c13", dmaapConsumerCpeAuthenticationConfig.consumerGroup()),
-            () -> assertEquals(Integer.valueOf(5), dmaapConsumerCpeAuthenticationConfig.timeoutMs()),
-            () -> assertEquals(Integer.valueOf(10), dmaapConsumerCpeAuthenticationConfig.messageLimit())
+            () -> assertEquals("we-are-message-router2.us",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapHostName()),
+            () -> assertEquals(3902,
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapPortNumber()),
+            () -> assertEquals("unauthenticated.CPE_AUTHENTICATION",
+                        configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapTopicName()),
+            () -> assertEquals("https",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapProtocol()),
+            () -> assertEquals("some-user",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapUserName()),
+            () -> assertEquals("some-password",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getDmaapContentType()),
+            () -> assertEquals("c13",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getConsumerId()),
+            () -> assertEquals("OpenDcae-c13",
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getConsumerGroup()),
+            () -> assertEquals(5, configuration.getDmaapCpeAuthenticationConsumerProperties().getTimeoutMs()),
+            () -> assertEquals(10,
+                    configuration.getDmaapCpeAuthenticationConsumerProperties().getMessageLimit())
         );
 
-        DmaapPublisherConfiguration dmaapPublisherConfiguration = configuration.getDmaapPublisherConfiguration();
         assertAll("DMaaP Publisher Configuration Properties",
-            () -> assertEquals("we-are-message-router3.us", dmaapPublisherConfiguration.dmaapHostName()),
-            () -> assertEquals(Integer.valueOf(3903), dmaapPublisherConfiguration.dmaapPortNumber()),
-            () -> assertEquals("events/unauthenticated.DCAE_CL_OUTPUT",
-                    dmaapPublisherConfiguration.dmaapTopicName()),
-            () -> assertEquals("https", dmaapPublisherConfiguration.dmaapProtocol()),
-            () -> assertEquals("some-user", dmaapPublisherConfiguration.dmaapUserName()),
-            () -> assertEquals("some-password", dmaapPublisherConfiguration.dmaapUserPassword()),
-            () -> assertEquals("application/json", dmaapPublisherConfiguration.dmaapContentType())
+            () -> assertEquals("we-are-message-router3.us",
+                    configuration.getDmaapProducerProperties().getDmaapHostName()),
+            () -> assertEquals(3903, configuration.getDmaapProducerProperties().getDmaapPortNumber()),
+            () -> assertEquals("unauthenticated.DCAE_CL_OUTPUT",
+                        configuration.getDmaapProducerProperties().getDmaapTopicName()),
+            () -> assertEquals("https", configuration.getDmaapProducerProperties().getDmaapProtocol()),
+            () -> assertEquals("some-user", configuration.getDmaapProducerProperties().getDmaapUserName()),
+            () -> assertEquals("some-password",
+                    configuration.getDmaapProducerProperties().getDmaapUserPassword()),
+            () -> assertEquals("application/json",
+                    configuration.getDmaapProducerProperties().getDmaapContentType())
         );
 
         assertAll("Generic Application Properties",
@@ -391,12 +423,11 @@ class ApplicationConfigurationTest {
 
         assertAll("Security Application Properties",
             () -> assertTrue(aaiClientConfiguration.enableAaiCertAuth()),
-            () -> assertTrue(dmaapConsumerReRegistrationConfig.enableDmaapCertAuth()),
-            () -> assertEquals("test key store path - update", aaiClientConfiguration.keyStorePath()),
-            () -> assertEquals("test key store password path - update",
+            () -> assertEquals("KeyStore-update.jks", aaiClientConfiguration.keyStorePath()),
+            () -> assertEquals("KeyStorePass-update.txt",
                         aaiClientConfiguration.keyStorePasswordPath()),
-            () -> assertEquals("test trust store path - update", aaiClientConfiguration.trustStorePath()),
-            () -> assertEquals("test trust store password path - update",
+            () -> assertEquals("KeyStore-update.jks", aaiClientConfiguration.trustStorePath()),
+            () -> assertEquals("KeyStorePass-update.txt",
                         aaiClientConfiguration.trustStorePasswordPath())
         );
     }
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ConsulConfigurationGatewayTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ConsulConfigurationGatewayTest.java
index 1acf864d..1d1bce2a 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ConsulConfigurationGatewayTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ConsulConfigurationGatewayTest.java
@@ -48,13 +48,13 @@ class ConsulConfigurationGatewayTest {
     }
 
     ConsulConfigurationGatewayTest() {
-        ApplicationConfiguration configuration = Mockito.mock(ApplicationConfiguration.class);
+        var configuration = Mockito.mock(ApplicationConfiguration.class);
         this.configurationGateway = new ConsulConfigurationGateway(configuration);
     }
 
     @Test
     void passingValidJson_constructsGeneratedAppConfigObject() {
-        final String validJson = "{"
+        final var validJson = "{"
                 + "\"dmaap.protocol\": \"http\","
                 + "\"dmaap.contentType\": \"application/json\","
                 + "\"dmaap.consumer.consumerId\": \"c12\","
@@ -219,7 +219,7 @@ class ConsulConfigurationGatewayTest {
                 .streamPublishesMap(Collections.singletonMap("config_key_3", streamsObject3))
                 .build();
 
-        ConsulConfigurationGateway spiedGateway = Mockito.spy(configurationGateway);
+        var spiedGateway = Mockito.spy(configurationGateway);
         doReturn(false).when(spiedGateway).environmentNotReady();
         assertEquals(expectedConfiguration,
                 spiedGateway.generateAppConfigObject(jsonParser.parse(validJson).getAsJsonObject()));
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/controllers/BbsEventProcessorControllerTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/controllers/BbsEventProcessorControllerTest.java
index bacb6c3e..8cc7d5cc 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/controllers/BbsEventProcessorControllerTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/controllers/BbsEventProcessorControllerTest.java
@@ -42,7 +42,6 @@ import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest;
 import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.web.servlet.MockMvc;
-import org.springframework.test.web.servlet.MvcResult;
 
 @WebMvcTest(BbsEventProcessorController.class)
 @DisplayName("BBS Event Processor Controllers MVC Unit-Tests")
@@ -67,7 +66,7 @@ class BbsEventProcessorControllerTest {
 
     @Test
     void sendingHeartBeatRestCall_RespondsWithAlive() throws Exception {
-        MvcResult heartBeatResult = mockMvc.perform(get("/heartbeat")).andReturn();
+        var heartBeatResult = mockMvc.perform(get("/heartbeat")).andReturn();
 
         mockMvc.perform(asyncDispatch(heartBeatResult))
                 .andExpect(status().isOk())
@@ -76,7 +75,7 @@ class BbsEventProcessorControllerTest {
 
     @Test
     void sendingReRegistrationSubmissionRestCall_RespondsWithOk() throws Exception {
-        MvcResult reregistrationSubmissionResult = mockMvc.perform(post("/poll-reregistration-events")).andReturn();
+        var reregistrationSubmissionResult = mockMvc.perform(post("/poll-reregistration-events")).andReturn();
 
         mockMvc.perform(asyncDispatch(reregistrationSubmissionResult))
                 .andExpect(status().isOk())
@@ -86,7 +85,7 @@ class BbsEventProcessorControllerTest {
 
     @Test
     void sendingCpeAuthenticationSubmissionRestCall_RespondsWithOk() throws Exception {
-        MvcResult reregistrationSubmissionResult = mockMvc.perform(post("/poll-cpe-authentication-events")).andReturn();
+        var reregistrationSubmissionResult = mockMvc.perform(post("/poll-cpe-authentication-events")).andReturn();
 
         mockMvc.perform(asyncDispatch(reregistrationSubmissionResult))
                 .andExpect(status().isOk())
@@ -97,7 +96,7 @@ class BbsEventProcessorControllerTest {
     @Test
     void sendingStartTasksRestCall_ifItSucceeds_RespondsWithOk() throws Exception {
         when(scheduler.reScheduleProcessingTasks()).thenReturn(true);
-        MvcResult startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
+        var startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
 
         mockMvc.perform(asyncDispatch(startTasksResult))
                 .andExpect(status().isOk())
@@ -108,7 +107,7 @@ class BbsEventProcessorControllerTest {
     @Test
     void sendingStartTasksRestCall_ifItFails_RespondsWithNotAcceptable() throws Exception {
         when(scheduler.reScheduleProcessingTasks()).thenReturn(false);
-        MvcResult startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
+        var startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
 
         mockMvc.perform(asyncDispatch(startTasksResult))
                 .andExpect(status().isNotAcceptable())
@@ -119,7 +118,7 @@ class BbsEventProcessorControllerTest {
     @Test
     void sendingCancelTasksRestCall_ifItSucceeds_RespondsWithOk() throws Exception {
         when(scheduler.cancelScheduledProcessingTasks()).thenReturn(true);
-        MvcResult cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
+        var cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
 
         mockMvc.perform(asyncDispatch(cancellationResult))
                 .andExpect(status().isOk())
@@ -130,7 +129,7 @@ class BbsEventProcessorControllerTest {
     @Test
     void sendingCancelTasksRestCall_ifItFails_RespondsWithNotAcceptable() throws Exception {
         when(scheduler.cancelScheduledProcessingTasks()).thenReturn(false);
-        MvcResult cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
+        var cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
 
         mockMvc.perform(asyncDispatch(cancellationResult))
                 .andExpect(status().isNotAcceptable())
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/model/GsonSerializationsTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/model/GsonSerializationsTest.java
index fd43b8be..30cc4ab8 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/model/GsonSerializationsTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/model/GsonSerializationsTest.java
@@ -22,7 +22,6 @@ package org.onap.bbs.event.processor.model;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.TypeAdapterFactory;
 
@@ -42,13 +41,13 @@ class GsonSerializationsTest {
     @Test
     void creatingReRegistrationJsonBody_returnsJsonInString() {
 
-        String correlationId = "NokiaCorrelationId";
-        String attachmentPoint = "olt2/1/1";
-        String remoteId = "RemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var correlationId = "NokiaCorrelationId";
+        var attachmentPoint = "olt2/1/1";
+        var remoteId = "RemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String template = "{"
+        var template = "{"
                 + "\"correlationId\":\"%s\","
                 + "\"attachment-point\":\"%s\","
                 + "\"remote-id\":\"%s\","
@@ -65,7 +64,7 @@ class GsonSerializationsTest {
                 .build();
 
 
-        String expectedResult = String.format(template, correlationId, attachmentPoint, remoteId, cvlan, svlan);
+        var expectedResult = String.format(template, correlationId, attachmentPoint, remoteId, cvlan, svlan);
 
         assertEquals(expectedResult, new ReRegistrationJsonBodyBuilder().createJsonBody(model));
     }
@@ -73,14 +72,14 @@ class GsonSerializationsTest {
     @Test
     void creatingCpeAuthenticationJsonBody_returnsJsonInString() {
 
-        String correlationId = "NokiaCorrelationID";
-        AuthenticationState oldAuthenticationState = AuthenticationState.IN_SERVICE;
-        AuthenticationState newAuthenticationState = AuthenticationState.OUT_OF_SERVICE;
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var correlationId = "NokiaCorrelationID";
+        var oldAuthenticationState = AuthenticationState.IN_SERVICE;
+        var newAuthenticationState = AuthenticationState.OUT_OF_SERVICE;
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String template = "{"
+        var template = "{"
                 + "\"correlationId\":\"%s\","
                 + "\"old-authentication-state\":\"%s\","
                 + "\"new-authentication-state\":\"%s\","
@@ -99,7 +98,7 @@ class GsonSerializationsTest {
                 .build();
 
 
-        String expectedResult = String.format(template, correlationId, oldAuthenticationState.getNameInOnap(),
+        var expectedResult = String.format(template, correlationId, oldAuthenticationState.getNameInOnap(),
                 newAuthenticationState.getNameInOnap(), stateInterface, rgwMacAddress, swVersion);
 
         assertEquals(expectedResult, new CpeAuthenticationJsonBodyBuilder().createJsonBody(model));
@@ -108,27 +107,27 @@ class GsonSerializationsTest {
     @Test
     void creatingDcaeControlLoopJsonBody_returnsJsonInString() {
 
-        String closedLoopEventClient = "DCAE.BBS_mSInstance";
-        String policyVersion = "1.0.0.5";
-        String policyName = "CPE_Authentication";
-        String policyScope =
+        var closedLoopEventClient = "DCAE.BBS_mSInstance";
+        var policyVersion = "1.0.0.5";
+        var policyName = "CPE_Authentication";
+        var policyScope =
                 "service=HSIAService,type=SampleType,"
                         + "closedLoopControlName=CL-CPE_A-d925ed73-8231-4d02-9545-db4e101f88f8";
-        String targetType = "VM";
-        long closedLoopAlarmStart = 1484677482204798L;
-        String closedLoopEventStatus = "ONSET";
-        String closedLoopControlName = "ControlLoop-CPE_A-2179b738-fd36-4843-a71a-a8c24c70c88b";
-        String version = "1.0.2";
-        String target = "vserver.vserver-name";
-        String requestId = "97964e10-686e-4790-8c45-bdfa61df770f";
-        String from = "DCAE";
+        var targetType = "VM";
+        var closedLoopAlarmStart = 1484677482204798L;
+        var closedLoopEventStatus = "ONSET";
+        var closedLoopControlName = "ControlLoop-CPE_A-2179b738-fd36-4843-a71a-a8c24c70c88b";
+        var version = "1.0.2";
+        var target = "vserver.vserver-name";
+        var requestId = "97964e10-686e-4790-8c45-bdfa61df770f";
+        var from = "DCAE";
 
         Map<String, String> aaiEnrichmentData = new LinkedHashMap<>();
         aaiEnrichmentData.put("service-information.service-instance-id", "service-instance-id-example");
         aaiEnrichmentData.put("cvlan-id", "example cvlan-id");
         aaiEnrichmentData.put("svlan-id", "example svlan-id");
 
-        String template = "{"
+        var template = "{"
                 + "\"closedLoopEventClient\":\"%s\","
                 + "\"policyVersion\":\"%s\","
                 + "\"policyName\":\"%s\","
@@ -165,7 +164,7 @@ class GsonSerializationsTest {
                 .originator(from)
                 .build();
 
-        String expectedResult = String.format(template,
+        var expectedResult = String.format(template,
                 closedLoopEventClient,
                 policyVersion,
                 policyName,
@@ -185,14 +184,14 @@ class GsonSerializationsTest {
     @Test
     void pnfAaiObject_IsSerializedSuccessfully() {
 
-        GsonBuilder gsonBuilder = new GsonBuilder();
+        var gsonBuilder = new GsonBuilder();
         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
-        Gson gson = gsonBuilder.create();
+        var gson = gsonBuilder.create();
 
-        String pnfName = "NokiaCorrelationID";
-        String swVersion = "1.2";
+        var pnfName = "NokiaCorrelationID";
+        var swVersion = "1.2";
 
-        String template = "{"
+        var template = "{"
                 + "\"pnf-name\":\"%s\","
                 + "\"in-maint\":true,"
                 + "\"sw-version\":\"%s\","
@@ -288,7 +287,7 @@ class GsonSerializationsTest {
                 .build();
 
 
-        String jsonPnfObject = String.format(template, pnfName, swVersion);
+        var jsonPnfObject = String.format(template, pnfName, swVersion);
 
         assertEquals(jsonPnfObject, gson.toJson(pnfAaiObject));
         assertEquals(pnfAaiObject, gson.fromJson(jsonPnfObject, ImmutablePnfAaiObject.class));
@@ -297,14 +296,14 @@ class GsonSerializationsTest {
     @Test
     void serviceInstanceAaiObject_IsSerializedSuccessfully() {
 
-        GsonBuilder gsonBuilder = new GsonBuilder();
+        var gsonBuilder = new GsonBuilder();
         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
-        Gson gson = gsonBuilder.create();
+        var gson = gsonBuilder.create();
 
-        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
-        String orchestrationStatus = "active";
+        var serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        var orchestrationStatus = "active";
 
-        String template = "{"
+        var template = "{"
                 + "\"service-instance-id\":\"%s\","
                 + "\"orchestration-status\":\"%s\","
                 + "\"relationship-list\":{"
@@ -370,7 +369,7 @@ class GsonSerializationsTest {
                 .build();
 
 
-        String jsonServiceInstanceObject = String.format(template, serviceInstanceId, orchestrationStatus);
+        var jsonServiceInstanceObject = String.format(template, serviceInstanceId, orchestrationStatus);
 
         assertEquals(jsonServiceInstanceObject, gson.toJson(serviceInstanceAaiObject));
         assertEquals(serviceInstanceAaiObject, gson.fromJson(jsonServiceInstanceObject,
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/CpeAuthenticationPipelineTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/CpeAuthenticationPipelineTest.java
index c4bef9df..50fcdefb 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/CpeAuthenticationPipelineTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/CpeAuthenticationPipelineTest.java
@@ -20,7 +20,7 @@
 
 package org.onap.bbs.event.processor.pipelines;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.times;
@@ -64,8 +64,7 @@ import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
 import org.onap.bbs.event.processor.tasks.AaiClientTask;
 import org.onap.bbs.event.processor.tasks.DmaapCpeAuthenticationConsumerTask;
 import org.onap.bbs.event.processor.tasks.DmaapPublisherTask;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
-import org.springframework.http.HttpStatus;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -80,12 +79,12 @@ class CpeAuthenticationPipelineTest {
     private DmaapPublisherTask publisherTask;
     private AaiClientTask aaiClientTask;
 
-    private HttpResponse httpResponse;
+    private MessageRouterPublishResponse publishResponse;
 
     @BeforeEach
     void setup() {
 
-        httpResponse = Mockito.mock(HttpResponse.class);
+        publishResponse = Mockito.mock(MessageRouterPublishResponse.class);
 
         configuration = Mockito.mock(ApplicationConfiguration.class);
         consumerTask = Mockito.mock(DmaapCpeAuthenticationConsumerTask.class);
@@ -148,12 +147,12 @@ class CpeAuthenticationPipelineTest {
     @Test
     void noResponseFromAai_PipelineTimesOut() throws SSLException {
 
-        String pnfName = "olt1";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
+        var pnfName = "olt1";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -181,13 +180,13 @@ class CpeAuthenticationPipelineTest {
     @Test
     void noResponseWhilePublishing_PipelineTimesOut() throws SSLException {
 
-        String pnfName = "olt1";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName = "olt1";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -199,12 +198,12 @@ class CpeAuthenticationPipelineTest {
                 .swVersion(swVersion)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
+        var pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, rgwMacAddress);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
@@ -217,7 +216,7 @@ class CpeAuthenticationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.never());
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.never());
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
@@ -230,13 +229,13 @@ class CpeAuthenticationPipelineTest {
     @Test
     void singleCorrectEvent_handleSuccessfully() throws SSLException {
 
-        String pnfName = "olt1";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName = "olt1";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -248,12 +247,12 @@ class CpeAuthenticationPipelineTest {
                 .swVersion(swVersion)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
+        var pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, rgwMacAddress);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -266,13 +265,13 @@ class CpeAuthenticationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
@@ -281,16 +280,16 @@ class CpeAuthenticationPipelineTest {
     @Test
     void twoCorrectEvents_handleSuccessfully() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress1 = "00:0a:95:8d:78:16";
-        final String rgwMacAddress2 = "00:0a:95:8d:78:17";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
-        String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress1 = "00:0a:95:8d:78:16";
+        var rgwMacAddress2 = "00:0a:95:8d:78:17";
+        var swVersion = "1.2";
+        var hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
+        var hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -310,19 +309,19 @@ class CpeAuthenticationPipelineTest {
                 .swVersion(swVersion)
                 .build();
 
-        PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, hsiCfsServiceInstanceId1);
-        PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, hsiCfsServiceInstanceId2);
-        ServiceInstanceAaiObject hsiCfsServiceInstance1 =
+        var pnfAaiObject1 = constructPnfObject(pnfName1, hsiCfsServiceInstanceId1);
+        var pnfAaiObject2 = constructPnfObject(pnfName2, hsiCfsServiceInstanceId2);
+        var hsiCfsServiceInstance1 =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, rgwMacAddress1);
-        ServiceInstanceAaiObject hsiCfsServiceInstance2 =
+        var hsiCfsServiceInstance2 =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, rgwMacAddress2);
 
         // Prepare Mocks
-        String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
-        String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
-        String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
+        var pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
+        var cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance1.getServiceInstanceId());
-        String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance2.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -339,14 +338,14 @@ class CpeAuthenticationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
                 .thenReturn(Mono.just(hsiCfsServiceInstance2));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(publisherTask, times(2)).execute(any(ControlLoopPublisherDmaapModel.class));
@@ -355,12 +354,12 @@ class CpeAuthenticationPipelineTest {
     @Test
     void singleEvent_withPnfErrorReply_handleGracefully() throws SSLException {
 
-        String pnfName = "olt1";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
+        var pnfName = "olt1";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -389,90 +388,16 @@ class CpeAuthenticationPipelineTest {
     }
 
     @Test
-    void twoEvents_FirstOk_SecondUnmatchedMac_handleCorrectOnly() throws SSLException {
-
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress1 = "00:0a:95:8d:78:16";
-        final String rgwMacAddress2 = "00:0a:95:8d:78:17";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
-        String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
-
-        // Prepare stubbed replies
-        CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
-                .correlationId(pnfName1)
-                .oldAuthenticationState(oldAuthenticationState)
-                .newAuthenticationState(newAuthenticationState)
-                .stateInterface(stateInterface)
-                .rgwMacAddress(rgwMacAddress1)
-                .swVersion(swVersion)
-                .build();
-        CpeAuthenticationConsumerDmaapModel secondEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
-                .correlationId(pnfName2)
-                .oldAuthenticationState(oldAuthenticationState)
-                .newAuthenticationState(newAuthenticationState)
-                .stateInterface(stateInterface)
-                .rgwMacAddress(rgwMacAddress2)
-                .swVersion(swVersion)
-                .build();
-
-        PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, hsiCfsServiceInstanceId1);
-        PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, hsiCfsServiceInstanceId2);
-        ServiceInstanceAaiObject hsiCfsServiceInstance1 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, rgwMacAddress1);
-        ServiceInstanceAaiObject hsiCfsServiceInstance2 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2,
-                        "Having unmatched RGW MAC address");
-
-        // Prepare Mocks
-        String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
-        String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
-        String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
-                hsiCfsServiceInstance1.getServiceInstanceId());
-        String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
-                hsiCfsServiceInstance2.getServiceInstanceId());
-
-        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
-        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME))
-                .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
-
-        when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl1)).thenReturn(Mono.just(pnfAaiObject1));
-        when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl2)).thenReturn(Mono.just(pnfAaiObject2));
-
-        when(aaiClientTask
-                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl1))
-                .thenReturn(Mono.just(hsiCfsServiceInstance1));
-        when(aaiClientTask
-                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
-                .thenReturn(Mono.just(hsiCfsServiceInstance2));
-
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
-
-        // Execute the pipeline
-        StepVerifier.create(pipeline.executePipeline())
-                .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
-                .verifyComplete();
-
-        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
-    }
-
-    @Test
     void twoEvents_firstOk_secondWithPnfErrorReply_handleCorrectOnly() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -492,12 +417,12 @@ class CpeAuthenticationPipelineTest {
                 .swVersion(swVersion)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName1, hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
+        var pnfAaiObject = constructPnfObject(pnfName1, hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, rgwMacAddress);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -510,13 +435,13 @@ class CpeAuthenticationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(aaiClientTask, times(2)).executePnfRetrieval(anyString(), anyString());
@@ -527,14 +452,14 @@ class CpeAuthenticationPipelineTest {
     @Test
     void twoEvents_firstWithPnfErrorReply_secondOk_handleCorrectOnly() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -554,12 +479,12 @@ class CpeAuthenticationPipelineTest {
                 .swVersion(swVersion)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName2, hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
+        var pnfAaiObject = constructPnfObject(pnfName2, hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, rgwMacAddress);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -572,13 +497,13 @@ class CpeAuthenticationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(aaiClientTask, times(2))
@@ -630,7 +555,7 @@ class CpeAuthenticationPipelineTest {
     private ServiceInstanceAaiObject constructHsiCfsServiceInstanceObject(String hsiCfsServiceInstanceId,
                                                                           String pnfName,
                                                                           String rgwMacAddress) {
-        String orchestrationStatus = "active";
+        var orchestrationStatus = "active";
 
         RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
                 ImmutableRelationshipEntryAaiObject.builder()
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/ReRegistrationPipelineTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/ReRegistrationPipelineTest.java
index 9453db3d..815cb5cc 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/ReRegistrationPipelineTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/ReRegistrationPipelineTest.java
@@ -20,7 +20,7 @@
 
 package org.onap.bbs.event.processor.pipelines;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.times;
@@ -64,8 +64,7 @@ import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
 import org.onap.bbs.event.processor.tasks.AaiClientTask;
 import org.onap.bbs.event.processor.tasks.DmaapPublisherTask;
 import org.onap.bbs.event.processor.tasks.DmaapReRegistrationConsumerTask;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
-import org.springframework.http.HttpStatus;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -80,12 +79,12 @@ class ReRegistrationPipelineTest {
     private DmaapPublisherTask publisherTask;
     private AaiClientTask aaiClientTask;
 
-    private HttpResponse httpResponse;
+    private MessageRouterPublishResponse publishResponse;
 
     @BeforeEach
     void setup() {
 
-        httpResponse = Mockito.mock(HttpResponse.class);
+        publishResponse = Mockito.mock(MessageRouterPublishResponse.class);
 
         configuration = Mockito.mock(ApplicationConfiguration.class);
         consumerTask = Mockito.mock(DmaapReRegistrationConsumerTask.class);
@@ -148,11 +147,11 @@ class ReRegistrationPipelineTest {
     @Test
     void noResponseFromAai_PipelineTimesOut() throws SSLException {
 
-        String pnfName = "olt1";
-        String attachmentPoint = "olt2-2-2";
-        String remoteId = "newRemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var pnfName = "olt1";
+        var attachmentPoint = "olt2-2-2";
+        var remoteId = "newRemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -179,12 +178,12 @@ class ReRegistrationPipelineTest {
     @Test
     void noResponseWhilePublishing_PipelineTimesOut() throws SSLException {
 
-        String pnfName = "olt1";
-        String attachmentPoint = "olt2-2-2";
-        String remoteId = "newRemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName = "olt1";
+        var attachmentPoint = "olt2-2-2";
+        var remoteId = "newRemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -195,12 +194,11 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "olt1-1-1", hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
+        var pnfAaiObject = constructPnfObject(pnfName, "olt1-1-1", hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
@@ -213,7 +211,7 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.never());
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.never());
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
@@ -226,12 +224,12 @@ class ReRegistrationPipelineTest {
     @Test
     void singleCorrectEvent_PnfHavingNoLogicalLink_handleGracefully() throws SSLException {
 
-        String pnfName = "olt1";
-        String attachmentPoint = "olt2-2-2";
-        String remoteId = "newRemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName = "olt1";
+        var attachmentPoint = "olt2-2-2";
+        var remoteId = "newRemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -242,12 +240,12 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObjectWithoutLogicalLink(pnfName, hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
+        var pnfAaiObject = constructPnfObjectWithoutLogicalLink(pnfName, hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance =
                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -260,8 +258,8 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
@@ -276,12 +274,12 @@ class ReRegistrationPipelineTest {
     @Test
     void singleCorrectEvent_handleSuccessfully() throws SSLException {
 
-        String pnfName = "olt1";
-        String attachmentPoint = "olt2-2-2";
-        String remoteId = "newRemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName = "olt1";
+        var attachmentPoint = "olt2-2-2";
+        var remoteId = "newRemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -292,12 +290,11 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "old-attachment-point", hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
+        var pnfAaiObject = constructPnfObject(pnfName, "old-attachment-point", hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -310,13 +307,13 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
@@ -325,17 +322,17 @@ class ReRegistrationPipelineTest {
     @Test
     void twoCorrectEvents_handleSuccessfully() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        String attachmentPoint1 = "olt1-1-1";
-        String attachmentPoint2 = "olt2-2-2";
-        String remoteId1 = "newRemoteId1";
-        String remoteId2 = "newRemoteId2";
-        String cvlan1 = "1005";
-        String cvlan2 = "1006";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
-        String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var attachmentPoint1 = "olt1-1-1";
+        var attachmentPoint2 = "olt2-2-2";
+        var remoteId1 = "newRemoteId1";
+        var remoteId2 = "newRemoteId2";
+        var cvlan1 = "1005";
+        var cvlan2 = "1006";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
+        var hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -353,19 +350,17 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
-        PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, "olt2-2-0", hsiCfsServiceInstanceId2);
-        ServiceInstanceAaiObject hsiCfsServiceInstance1 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
-        ServiceInstanceAaiObject hsiCfsServiceInstance2 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
+        var pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
+        var pnfAaiObject2 = constructPnfObject(pnfName2, "olt2-2-0", hsiCfsServiceInstanceId2);
+        var hsiCfsServiceInstance1 = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
+        var hsiCfsServiceInstance2 = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
 
         // Prepare Mocks
-        String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
-        String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
-        String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
+        var pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
+        var cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance1.getServiceInstanceId());
-        String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance2.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -382,14 +377,14 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
                 .thenReturn(Mono.just(hsiCfsServiceInstance2));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(publisherTask, times(2)).execute(any(ControlLoopPublisherDmaapModel.class));
@@ -398,11 +393,11 @@ class ReRegistrationPipelineTest {
     @Test
     void singleEvent_withPnfErrorReply_handleGracefully() throws SSLException {
 
-        String pnfName = "olt1";
-        String attachmentPoint = "olt2-2-2";
-        String remoteId = "newRemoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var pnfName = "olt1";
+        var attachmentPoint = "olt2-2-2";
+        var remoteId = "newRemoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -432,17 +427,17 @@ class ReRegistrationPipelineTest {
     @Test
     void twoEvents_FirstOk_SecondNotRelocation_handleCorrectOnly() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        String attachmentPoint1 = "olt1-1-1";
-        String attachmentPoint2 = "olt2-2-2";
-        String remoteId1 = "newRemoteId1";
-        String remoteId2 = "newRemoteId2";
-        String cvlan1 = "1005";
-        String cvlan2 = "1006";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
-        String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var attachmentPoint1 = "olt1-1-1";
+        var attachmentPoint2 = "olt2-2-2";
+        var remoteId1 = "newRemoteId1";
+        var remoteId2 = "newRemoteId2";
+        var cvlan1 = "1005";
+        var cvlan2 = "1006";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
+        var hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -460,19 +455,17 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
-        PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, attachmentPoint2, hsiCfsServiceInstanceId2);
-        ServiceInstanceAaiObject hsiCfsServiceInstance1 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
-        ServiceInstanceAaiObject hsiCfsServiceInstance2 =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
+        var pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
+        var pnfAaiObject2 = constructPnfObject(pnfName2, attachmentPoint2, hsiCfsServiceInstanceId2);
+        var hsiCfsServiceInstance1 = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
+        var hsiCfsServiceInstance2 = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
 
         // Prepare Mocks
-        String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
-        String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
-        String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
+        var pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
+        var cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance1.getServiceInstanceId());
-        String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance2.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -489,13 +482,13 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
                 .thenReturn(Mono.just(hsiCfsServiceInstance2));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
@@ -504,16 +497,16 @@ class ReRegistrationPipelineTest {
     @Test
     void twoEvents_firstOk_secondWithPnfErrorReply_handleCorrectOnly() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        String attachmentPoint1 = "olt1-1-1";
-        String attachmentPoint2 = "olt2-2-2";
-        String remoteId1 = "newRemoteId1";
-        String remoteId2 = "newRemoteId2";
-        String cvlan1 = "1005";
-        String cvlan2 = "1006";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var attachmentPoint1 = "olt1-1-1";
+        var attachmentPoint2 = "olt2-2-2";
+        var remoteId1 = "newRemoteId1";
+        var remoteId2 = "newRemoteId2";
+        var cvlan1 = "1005";
+        var cvlan2 = "1006";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -531,12 +524,11 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName1, "old-attachment-point", hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, cvlan1);
+        var pnfAaiObject = constructPnfObject(pnfName1, "old-attachment-point", hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, cvlan1);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -549,13 +541,13 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(aaiClientTask, times(2)).executePnfRetrieval(anyString(), anyString());
@@ -566,16 +558,16 @@ class ReRegistrationPipelineTest {
     @Test
     void twoEvents_firstWithPnfErrorReply_secondOk_handleCorrectOnly() throws SSLException {
 
-        String pnfName1 = "olt1";
-        String pnfName2 = "olt2";
-        String attachmentPoint1 = "olt1-1-1";
-        String attachmentPoint2 = "olt2-2-2";
-        String remoteId1 = "newRemoteId1";
-        String remoteId2 = "newRemoteId2";
-        String cvlan1 = "1005";
-        String cvlan2 = "1006";
-        String svlan = "100";
-        String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
+        var pnfName1 = "olt1";
+        var pnfName2 = "olt2";
+        var attachmentPoint1 = "olt1-1-1";
+        var attachmentPoint2 = "olt2-2-2";
+        var remoteId1 = "newRemoteId1";
+        var remoteId2 = "newRemoteId2";
+        var cvlan1 = "1005";
+        var cvlan2 = "1006";
+        var svlan = "100";
+        var hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 
         // Prepare stubbed replies
         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
@@ -593,12 +585,11 @@ class ReRegistrationPipelineTest {
                 .sVlan(svlan)
                 .build();
 
-        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName2, "old-attachment-point", hsiCfsServiceInstanceId);
-        ServiceInstanceAaiObject hsiCfsServiceInstance =
-                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, cvlan2);
+        var pnfAaiObject = constructPnfObject(pnfName2, "old-attachment-point", hsiCfsServiceInstanceId);
+        var hsiCfsServiceInstance = constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, cvlan2);
 
         // Prepare Mocks
-        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+        var cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
                 hsiCfsServiceInstance.getServiceInstanceId());
 
         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
@@ -611,13 +602,13 @@ class ReRegistrationPipelineTest {
                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 
-        when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
-        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
+        when(publishResponse.successful()).thenReturn(true);
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Flux.just(publishResponse));
 
         // Execute the pipeline
         StepVerifier.create(pipeline.executePipeline())
                 .expectSubscription()
-                .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
+                .assertNext(r -> assertTrue(r.successful()))
                 .verifyComplete();
 
         verify(aaiClientTask, times(2))
@@ -719,7 +710,7 @@ class ReRegistrationPipelineTest {
     private ServiceInstanceAaiObject constructHsiCfsServiceInstanceObject(String hsiCfsServiceInstanceId,
                                                                              String pnfName,
                                                                              String cvlan) {
-        String orchestrationStatus = "active";
+        var orchestrationStatus = "active";
 
         RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
                 ImmutableRelationshipEntryAaiObject.builder()
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/SchedulerTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/SchedulerTest.java
index f721ca7e..b037e2cb 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/SchedulerTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/SchedulerTest.java
@@ -50,9 +50,9 @@ class SchedulerTest {
     SchedulerTest() {
         configuration = Mockito.mock(ApplicationConfiguration.class);
         taskScheduler = Mockito.mock(TaskScheduler.class);
-        ReRegistrationPipeline reRegistrationPipeline = Mockito.mock(ReRegistrationPipeline.class);
-        CpeAuthenticationPipeline cpeAuthenticationPipeline = Mockito.mock(CpeAuthenticationPipeline.class);
-        ConsulConfigurationGateway configurationGateway = Mockito.mock(ConsulConfigurationGateway.class);
+        var reRegistrationPipeline = Mockito.mock(ReRegistrationPipeline.class);
+        var cpeAuthenticationPipeline = Mockito.mock(CpeAuthenticationPipeline.class);
+        var configurationGateway = Mockito.mock(ConsulConfigurationGateway.class);
         this.applicationScheduler = new Scheduler(configuration, configurationGateway, taskScheduler,
                 reRegistrationPipeline, cpeAuthenticationPipeline);
     }
@@ -60,7 +60,7 @@ class SchedulerTest {
     @Test
     void scheduleTasksWithValidSchedulingPeriod_Succeeds() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
-        ScheduledFuture scheduledFuture = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture = Mockito.mock(ScheduledFuture.class);
         when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any(Instant.class), any(Duration.class)))
                .thenReturn(scheduledFuture);
 
@@ -75,8 +75,8 @@ class SchedulerTest {
     @Test
     void cancellingRunningTasksSucceeds_tasksAreDeleted() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
-        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
         when(scheduledFuture1.cancel(false)).thenReturn(true);
         when(scheduledFuture2.cancel(false)).thenReturn(true);
         when(scheduledFuture1.isCancelled()).thenReturn(true);
@@ -85,7 +85,7 @@ class SchedulerTest {
                 .thenReturn(scheduledFuture1).thenReturn(scheduledFuture2);
 
         applicationScheduler.setupScheduler();
-        boolean result = applicationScheduler.cancelScheduledProcessingTasks();
+        var result = applicationScheduler.cancelScheduledProcessingTasks();
         assertAll("Successfully cancelling tasks",
             () -> assertTrue(result, "Result of cancellation task"),
             () -> assertEquals(0, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
@@ -97,8 +97,8 @@ class SchedulerTest {
     @Test
     void cancellingRunningTasksPartiallyFailing_tasksAreNotDeleted() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
-        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
         when(scheduledFuture1.cancel(false)).thenReturn(true);
         when(scheduledFuture2.cancel(false)).thenReturn(false);
         when(scheduledFuture1.isCancelled()).thenReturn(true);
@@ -107,7 +107,7 @@ class SchedulerTest {
                 .thenReturn(scheduledFuture1).thenReturn(scheduledFuture2);
 
         applicationScheduler.setupScheduler();
-        boolean result = applicationScheduler.cancelScheduledProcessingTasks();
+        var result = applicationScheduler.cancelScheduledProcessingTasks();
         assertAll("Partially cancelling tasks",
             () -> assertFalse(result, "Result of cancellation task"),
             () -> assertEquals(1, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
@@ -119,8 +119,8 @@ class SchedulerTest {
     @Test
     void cancellingRunningTasksFailingForAllOfThem_noTasksAreDeleted() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
-        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
         when(scheduledFuture1.cancel(false)).thenReturn(false);
         when(scheduledFuture2.cancel(false)).thenReturn(false);
         when(scheduledFuture1.isCancelled()).thenReturn(false);
@@ -129,7 +129,7 @@ class SchedulerTest {
                 .thenReturn(scheduledFuture1).thenReturn(scheduledFuture2);
 
         applicationScheduler.setupScheduler();
-        boolean result = applicationScheduler.cancelScheduledProcessingTasks();
+        var result = applicationScheduler.cancelScheduledProcessingTasks();
         assertAll("Failing in cancelling tasks",
             () -> assertFalse(result, "Result of cancellation task"),
             () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
@@ -141,15 +141,15 @@ class SchedulerTest {
     @Test
     void reSchedulingWithExistingActiveTasks_Fails() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
-        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
         when(scheduledFuture1.isCancelled()).thenReturn(false);
         when(scheduledFuture2.isCancelled()).thenReturn(false);
         when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any(Instant.class), any(Duration.class)))
                 .thenReturn(scheduledFuture1).thenReturn(scheduledFuture2);
 
         applicationScheduler.setupScheduler();
-        boolean result = applicationScheduler.reScheduleProcessingTasks();
+        var result = applicationScheduler.reScheduleProcessingTasks();
         assertAll("Rescheduling with active tasks",
             () -> assertFalse(result, "Result of re-scheduling"),
             () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
@@ -162,11 +162,11 @@ class SchedulerTest {
     void reSchedulingWithExistingCancelledTasks_Succeeds() {
         when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
         // Initial tasks
-        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
         // Re-scheduled tasks
-        ScheduledFuture scheduledFuture3 = Mockito.mock(ScheduledFuture.class);
-        ScheduledFuture scheduledFuture4 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture3 = Mockito.mock(ScheduledFuture.class);
+        var scheduledFuture4 = Mockito.mock(ScheduledFuture.class);
         when(scheduledFuture1.isCancelled()).thenReturn(true);
         when(scheduledFuture2.isCancelled()).thenReturn(true);
         when(scheduledFuture3.isCancelled()).thenReturn(false);
@@ -178,7 +178,7 @@ class SchedulerTest {
                 .thenReturn(scheduledFuture4);
 
         applicationScheduler.setupScheduler();
-        boolean result = applicationScheduler.reScheduleProcessingTasks();
+        var result = applicationScheduler.reScheduleProcessingTasks();
         assertAll("Rescheduling with cancelled tasks",
             () -> assertTrue(result, "Result of re-scheduling"),
             () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/AaiClientTaskImplTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/AaiClientTaskImplTest.java
index db5f7cb1..e13b8bc1 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/AaiClientTaskImplTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/AaiClientTaskImplTest.java
@@ -64,7 +64,7 @@ class AaiClientTaskImplTest {
 
     @BeforeEach
     void init() {
-        GsonBuilder gsonBuilder = new GsonBuilder();
+        var gsonBuilder = new GsonBuilder();
         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
         reactiveClient = Mockito.mock(AaiReactiveClient.class);
         task = new AaiClientTaskImpl(reactiveClient);
@@ -73,7 +73,7 @@ class AaiClientTaskImplTest {
     @Test
     void passingEmptyPnfObject_NothingHappens() throws AaiTaskException {
         when(reactiveClient.getPnfObjectDataFor(any(String.class))).thenReturn(Mono.empty());
-        Mono<PnfAaiObject> pnf = task.executePnfRetrieval("Empty PNF task", "some-url");
+        var pnf = task.executePnfRetrieval("Empty PNF task", "some-url");
 
         verify(reactiveClient).getPnfObjectDataFor("some-url");
         assertNull(pnf.block(Duration.ofSeconds(5)));
@@ -82,7 +82,7 @@ class AaiClientTaskImplTest {
     @Test
     void passingEmptyServiceInstanceObject_NothingHappens() throws AaiTaskException {
         when(reactiveClient.getServiceInstanceObjectDataFor(any(String.class))).thenReturn(Mono.empty());
-        Mono<ServiceInstanceAaiObject> serviceInstance =
+        var serviceInstance =
                 task.executeServiceInstanceRetrieval("Empty Service Instance task", "some-url");
 
         verify(reactiveClient).getServiceInstanceObjectDataFor("some-url");
@@ -92,8 +92,8 @@ class AaiClientTaskImplTest {
     @Test
     void passingPnfObject_taskSucceeds() throws AaiTaskException {
 
-        String pnfName = "pnf-1";
-        String attachmentPoint = "olt1-1-1";
+        var pnfName = "pnf-1";
+        var attachmentPoint = "olt1-1-1";
 
         // Build Relationship Data
         RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
@@ -142,7 +142,7 @@ class AaiClientTaskImplTest {
                 .build();
 
         when(reactiveClient.getPnfObjectDataFor(any(String.class))).thenReturn(Mono.just(pnfAaiObject));
-        Mono<PnfAaiObject> pnf = task.executePnfRetrieval("Normal PNF retrieval task", "some-url");
+        var pnf = task.executePnfRetrieval("Normal PNF retrieval task", "some-url");
 
         verify(reactiveClient).getPnfObjectDataFor("some-url");
         assertNotNull(pnf.block(Duration.ofSeconds(5)));
@@ -151,7 +151,7 @@ class AaiClientTaskImplTest {
                 .expectSubscription()
                 .consumeNextWith(aPnf -> {
                     Assertions.assertEquals(pnfName, aPnf.getPnfName(), "PNF Name in response does not match");
-                    String extractedAttachmentPoint = aPnf.getRelationshipListAaiObject().getRelationshipEntries()
+                    var extractedAttachmentPoint = aPnf.getRelationshipListAaiObject().getRelationshipEntries()
                             .stream()
                             .filter(e -> e.getRelatedTo().equals("logical-link"))
                             .flatMap(e -> e.getRelationshipData().stream())
@@ -167,8 +167,8 @@ class AaiClientTaskImplTest {
     @Test
     void passingServiceInstanceObject_taskSucceeds() throws AaiTaskException {
 
-        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
-        String orchestrationStatus = "active";
+        var serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        var orchestrationStatus = "active";
 
         // Build Relationship Data
         RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
@@ -207,7 +207,7 @@ class AaiClientTaskImplTest {
 
         when(reactiveClient.getServiceInstanceObjectDataFor(any(String.class)))
                 .thenReturn(Mono.just(serviceInstanceAaiObject));
-        Mono<ServiceInstanceAaiObject> serviceInstance =
+        var serviceInstance =
                 task.executeServiceInstanceRetrieval("Normal Service Instance retrieval task",
                         "some-url");
 
@@ -220,10 +220,10 @@ class AaiClientTaskImplTest {
                     Assertions.assertEquals(serviceInstanceId, instance.getServiceInstanceId(),
                             "Service Instance ID in response does not match");
 
-                    MetadataListAaiObject extractedMetadataListObject =
+                    var extractedMetadataListObject =
                             instance.getMetadataListAaiObject().orElseThrow(AaiClientTaskTestException::new);
 
-                    MetadataListAaiObject.MetadataEntryAaiObject extractedMetadataEntry =
+                    var extractedMetadataEntry =
                             extractedMetadataListObject.getMetadataEntries()
                                     .stream()
                                     .filter(m -> m.getMetaname().equals("cvlan"))
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapCpeAuthenticationConsumerTaskImplTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapCpeAuthenticationConsumerTaskImplTest.java
index 40bcb65d..f644d443 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapCpeAuthenticationConsumerTaskImplTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapCpeAuthenticationConsumerTaskImplTest.java
@@ -20,38 +20,39 @@
 
 package org.onap.bbs.event.processor.tasks;
 
-import static org.mockito.Mockito.doReturn;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.reset;
+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 com.google.gson.Gson;
-import com.google.gson.JsonElement;
-
-import java.util.Optional;
-
-import javax.net.ssl.SSLException;
 
 import org.junit.Assert;
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.DmaapCpeAuthenticationConsumerProperties;
 import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
 import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
 import org.onap.bbs.event.processor.model.ImmutableCpeAuthenticationConsumerDmaapModel;
 import org.onap.bbs.event.processor.utilities.CpeAuthenticationDmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapConsumerConfiguration;
-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 org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
 
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 class DmaapCpeAuthenticationConsumerTaskImplTest {
 
+    private static final String DMAAP_PROTOCOL = "http";
+    private static final String DMAAP_HOST = "message-router.onap.svc.cluster.local";
+    private static final int DMAAP_PORT = 3904;
+    private static final String DMAAP_TOPIC = "unauthenticated.CPE_AUTHENTICATION";
+    private static final String SUBSCRIBER_ID = "subscriberID";
+    private static final String SUBSCRIBER_GROUP = "subscriberGroup";
+
     private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE = "{\"event\": {"
             + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
             + "\"stateChangeFields\": {"
@@ -65,32 +66,34 @@ class DmaapCpeAuthenticationConsumerTaskImplTest {
 
     private static DmaapCpeAuthenticationConsumerTask dmaapConsumerTask;
     private static CpeAuthenticationConsumerDmaapModel cpeAuthenticationConsumerDmaapModel;
-    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
     private static String eventsArray;
+    private static MessageRouterSubscriber subscriber;
     private static Gson gson = new Gson();
+    private static ApplicationConfiguration configuration;
 
     @BeforeAll
-    static void setUp() throws SSLException {
-
-        final String sourceName = "PNF-CorrelationId";
-        final String oldAuthenticationState = "outOfService";
-        final String newAuthenticationState = "inService";
-        final String stateInterface = "stateInterface";
-        final String rgwMacAddress = "00:0a:95:8d:78:16";
-        final String swVersion = "1.2";
+    static void setUp() {
 
         // Mock Re-registration configuration
-        DmaapConsumerConfiguration dmaapConsumerConfiguration = testVersionOfDmaapConsumerConfiguration();
-        ApplicationConfiguration configuration = mock(ApplicationConfiguration.class);
-        when(configuration.getDmaapCpeAuthenticationConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
-
-        // Mock reactive DMaaP client
-        ConsumerReactiveHttpClientFactory httpClientFactory = mock(ConsumerReactiveHttpClientFactory.class);
-        dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
-        doReturn(dMaaPConsumerReactiveHttpClient).when(httpClientFactory).create(dmaapConsumerConfiguration);
-
-        dmaapConsumerTask = new DmaapCpeAuthenticationConsumerTaskImpl(configuration,
-                new CpeAuthenticationDmaapConsumerJsonParser(), httpClientFactory);
+        configuration = mock(ApplicationConfiguration.class);
+        var props = mock(DmaapCpeAuthenticationConsumerProperties.class);
+        when(props.getDmaapProtocol()).thenReturn(DMAAP_PROTOCOL);
+        when(props.getDmaapHostName()).thenReturn(DMAAP_HOST);
+        when(props.getDmaapPortNumber()).thenReturn(DMAAP_PORT);
+        when(props.getDmaapTopicName()).thenReturn(DMAAP_TOPIC);
+        when(props.getConsumerId()).thenReturn(SUBSCRIBER_ID);
+        when(props.getConsumerGroup()).thenReturn(SUBSCRIBER_GROUP);
+        when(configuration.getDmaapCpeAuthenticationConsumerProperties()).thenReturn(props);
+
+        var subscriberConfig = mock(MessageRouterSubscriberConfig.class);
+        when(configuration.getDmaapCpeAuthenticationConsumerConfiguration()).thenReturn(subscriberConfig);
+
+        var sourceName = "PNF-CorrelationId";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
         cpeAuthenticationConsumerDmaapModel = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
                 .correlationId(sourceName)
@@ -101,58 +104,42 @@ class DmaapCpeAuthenticationConsumerTaskImplTest {
                 .swVersion(swVersion)
                 .build();
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName, oldAuthenticationState,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName, oldAuthenticationState,
                 newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
         eventsArray = "[" + event + "]";
     }
 
-    @AfterEach
-    void resetMock() {
-        reset(dMaaPConsumerReactiveHttpClient);
-    }
-
     @Test
     void passingEmptyMessage_NothingHappens() throws Exception {
-        JsonElement empty = gson.toJsonTree("");
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty())).thenReturn(Mono.just(empty));
+        var empty = gson.toJsonTree("");
+        subscriber = mock(MessageRouterSubscriber.class);
+        when(subscriber.getElements(any())).thenReturn(Flux.just(empty));
+
+        dmaapConsumerTask = new DmaapCpeAuthenticationConsumerTaskImpl(configuration, subscriber,
+                new CpeAuthenticationDmaapConsumerJsonParser());
 
         StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
                 .expectSubscription()
                 .expectError(EmptyDmaapResponseException.class);
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
+
+        verify(subscriber, times(1)).getElements(any());
+        verifyNoMoreInteractions(subscriber);
     }
 
     @Test
     void passingNormalMessage_ResponseSucceeds() throws Exception {
-        JsonElement normalEventsArray = gson.toJsonTree(eventsArray);
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty()))
-                .thenReturn(Mono.just(normalEventsArray));
+        var normalEventsArray = gson.toJsonTree(eventsArray);
+        subscriber = mock(MessageRouterSubscriber.class);
+        when(subscriber.getElements(any())).thenReturn(Flux.just(normalEventsArray));
+
+        dmaapConsumerTask = new DmaapCpeAuthenticationConsumerTaskImpl(configuration, subscriber,
+                new CpeAuthenticationDmaapConsumerJsonParser());
 
         StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
                 .expectSubscription()
                 .consumeNextWith(e -> Assert.assertEquals(e, cpeAuthenticationConsumerDmaapModel));
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
-    }
-
-    private static DmaapConsumerConfiguration testVersionOfDmaapConsumerConfiguration() {
-        return new ImmutableDmaapConsumerConfiguration.Builder()
-                .consumerGroup("consumer-group")
-                .consumerId("consumer-id")
-                .dmaapContentType("application/json")
-                .dmaapHostName("message-router.onap.svc.cluster.local")
-                .dmaapPortNumber(3904)
-                .dmaapProtocol("http")
-                .dmaapUserName("admin")
-                .dmaapUserPassword("admin")
-                .trustStorePath("change it")
-                .trustStorePasswordPath("change_it")
-                .keyStorePath("change it")
-                .keyStorePasswordPath("change_it")
-                .enableDmaapCertAuth(false)
-                .dmaapTopicName("/events/unauthenticated.CPE_AUTHENTICATION")
-                .timeoutMs(-1)
-                .messageLimit(-1)
-                .build();
+        verify(subscriber, times(1)).getElements(any());
+        verifyNoMoreInteractions(subscriber);
     }
 }
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapPublisherTaskImplTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapPublisherTaskImplTest.java
index 436206d2..7c158489 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapPublisherTaskImplTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapPublisherTaskImplTest.java
@@ -21,7 +21,6 @@
 package org.onap.bbs.event.processor.tasks;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -30,55 +29,52 @@ import static org.mockito.Mockito.when;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.Optional;
-
-import javax.net.ssl.SSLException;
 
 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;
 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.DmaapProducerProperties;
 import org.onap.bbs.event.processor.exceptions.DmaapException;
 import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
 import org.onap.bbs.event.processor.model.ImmutableControlLoopPublisherDmaapModel;
-import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapPublisherConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.DMaaPPublisherReactiveHttpClient;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.producer.PublisherReactiveHttpClientFactory;
-import org.springframework.http.HttpStatus;
-
-import reactor.core.publisher.Mono;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterPublisher;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.MessageRouterPublishResponse;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterPublisherConfig;
+
+import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 class DmaapPublisherTaskImplTest {
 
+    private static final String DMAAP_PROTOCOL = "http";
+    private static final String DMAAP_HOST = "message-router.onap.svc.cluster.local";
+    private static final int DMAAP_PORT = 3904;
+    private static final String DMAAP_TOPIC = "unauthenticated.DCAE_CL_OUTPUT";
+
     private static ControlLoopPublisherDmaapModel controlLoopPublisherDmaapModel;
     private static DmaapPublisherTaskImpl task;
-    private static DMaaPPublisherReactiveHttpClient reactiveHttpClient;
     private static ApplicationConfiguration configuration;
-    private static DmaapPublisherConfiguration dmaapPublisherConfiguration;
 
     @BeforeAll
     static void setUp() {
-        dmaapPublisherConfiguration = testVersionOfDmaapPublisherConfiguration();
         configuration = mock(ApplicationConfiguration.class);
 
-        final String closedLoopEventClient = "DCAE.BBS_mSInstance";
-        final String policyVersion = "1.0.0.5";
-        final String policyName = "CPE_Authentication";
-        final String policyScope =
+        final var closedLoopEventClient = "DCAE.BBS_mSInstance";
+        final var policyVersion = "1.0.0.5";
+        final var policyName = "CPE_Authentication";
+        final var policyScope =
                 "service=HSIAService,type=SampleType,"
                         + "closedLoopControlName=CL-CPE_A-d925ed73-8231-4d02-9545-db4e101f88f8";
-        final String targetType = "VM";
-        final long closedLoopAlarmStart = 1484677482204798L;
-        final String closedLoopEventStatus = "ONSET";
-        final String closedLoopControlName = "ControlLoop-CPE_A-2179b738-fd36-4843-a71a-a8c24c70c88b";
-        final String version = "1.0.2";
-        final String target = "vserver.vserver-name";
-        final String requestId = "97964e10-686e-4790-8c45-bdfa61df770f";
-        final String from = "DCAE";
+        final var targetType = "VM";
+        final var closedLoopAlarmStart = 1484677482204798L;
+        final var closedLoopEventStatus = "ONSET";
+        final var closedLoopControlName = "ControlLoop-CPE_A-2179b738-fd36-4843-a71a-a8c24c70c88b";
+        final var version = "1.0.2";
+        final var target = "vserver.vserver-name";
+        final var requestId = "97964e10-686e-4790-8c45-bdfa61df770f";
+        final var from = "DCAE";
 
         final Map<String, String> aaiEnrichmentData = new LinkedHashMap<>();
         aaiEnrichmentData.put("service-information.service-instance-id", "service-instance-id-example");
@@ -100,77 +96,64 @@ class DmaapPublisherTaskImplTest {
                 .requestId(requestId)
                 .originator(from)
                 .build();
+        var props = mock(DmaapProducerProperties.class);
+        when(props.getDmaapProtocol()).thenReturn(DMAAP_PROTOCOL);
+        when(props.getDmaapHostName()).thenReturn(DMAAP_HOST);
+        when(props.getDmaapPortNumber()).thenReturn(DMAAP_PORT);
+        when(props.getDmaapTopicName()).thenReturn(DMAAP_TOPIC);
+        when(configuration.getDmaapProducerProperties()).thenReturn(props);
+
+        var publisherConfig = mock(MessageRouterPublisherConfig.class);
+        when(configuration.getDmaapPublisherConfiguration()).thenReturn(publisherConfig);
+
 
-        when(configuration.getDmaapPublisherConfiguration()).thenReturn(dmaapPublisherConfiguration);
     }
 
     @Test
     void passingNullMessage_ExceptionIsRaised() {
 
-        task = new DmaapPublisherTaskImpl(configuration);
-
         Executable executableFunction = () -> task.execute(null);
 
         Assertions.assertThrows(DmaapException.class, executableFunction, "Input message is invalid");
     }
 
     @Test
-    void passingNormalMessage_ReactiveClientProcessesIt() throws DmaapException, SSLException {
-        HttpResponse response = setupMocks(HttpStatus.OK.value());
-
-        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel)).expectSubscription()
-                .expectNext(response).verifyComplete();
-
-        verify(reactiveHttpClient, times(1))
-                .getDMaaPProducerResponse(controlLoopPublisherDmaapModel, Optional.empty());
-        verifyNoMoreInteractions(reactiveHttpClient);
-    }
+    void passingNormalMessage_ReactiveClientProcessesIt() throws DmaapException {
+        var publisher = mock(MessageRouterPublisher.class);
+        task = new DmaapPublisherTaskImpl(configuration, publisher);
 
-    @Test
-    void passingNormalMessage_IncorrectResponseIsHandled() throws DmaapException, SSLException {
-        HttpResponse response = setupMocks(HttpStatus.UNAUTHORIZED.value());
+        var response = mockResponse(true);
+        when(publisher.put(any(),any())).thenReturn(Flux.just(response));
 
-        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel)).expectSubscription()
-                .expectNext(response).verifyComplete();
+        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel))
+                .expectSubscription()
+                .assertNext(r -> Assertions.assertTrue(r.successful()))
+                .verifyComplete();
 
-        verify(reactiveHttpClient, times(1))
-                .getDMaaPProducerResponse(controlLoopPublisherDmaapModel, Optional.empty());
-        verifyNoMoreInteractions(reactiveHttpClient);
+        verify(publisher, times(1)).put(any(),any());
+        verifyNoMoreInteractions(publisher);
     }
 
-    // We can safely suppress unchecked assignment warning here since it is a mock class
-    @SuppressWarnings("unchecked")
-    private HttpResponse setupMocks(Integer httpResponseCode) throws SSLException {
-
-        HttpResponse response = mock(HttpResponse.class);
-        when(response.statusCode()).thenReturn(httpResponseCode);
-
-        reactiveHttpClient = mock(DMaaPPublisherReactiveHttpClient.class);
-        when(reactiveHttpClient.getDMaaPProducerResponse(any(), any(Optional.class)))
-                .thenReturn(Mono.just(response));
+    @Test
+    void passingNormalMessage_IncorrectResponseIsHandled() throws DmaapException {
+        var publisher = mock(MessageRouterPublisher.class);
+        task = new DmaapPublisherTaskImpl(configuration, publisher);
 
-        PublisherReactiveHttpClientFactory httpClientFactory = mock(PublisherReactiveHttpClientFactory.class);
-        doReturn(reactiveHttpClient).when(httpClientFactory).create(dmaapPublisherConfiguration);
+        var response = mockResponse(false);
+        when(publisher.put(any(),any())).thenReturn(Flux.just(response));
 
-        task = new DmaapPublisherTaskImpl(configuration, httpClientFactory);
+        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel))
+                .expectSubscription()
+                .assertNext(r -> Assertions.assertFalse(r.successful()))
+                .verifyComplete();
 
-        return response;
+        verify(publisher, times(1)).put(any(),any());
+        verifyNoMoreInteractions(publisher);
     }
 
-    private static DmaapPublisherConfiguration testVersionOfDmaapPublisherConfiguration() {
-        return new ImmutableDmaapPublisherConfiguration.Builder()
-                .dmaapContentType("application/json")
-                .dmaapHostName("message-router.onap.svc.cluster.local")
-                .dmaapPortNumber(3904)
-                .dmaapProtocol("http")
-                .dmaapUserName("admin")
-                .dmaapUserPassword("admin")
-                .trustStorePath("/opt/app/bbs/local/org.onap.bbs.trust.jks")
-                .trustStorePasswordPath("change_it")
-                .keyStorePath("/opt/app/bbs/local/org.onap.bbs.p12")
-                .keyStorePasswordPath("change_it")
-                .enableDmaapCertAuth(false)
-                .dmaapTopicName("/events/unauthenticated.DCAE_CL_OUTPUT")
-                .build();
+    private MessageRouterPublishResponse mockResponse(boolean isSuccess) {
+        var response = mock(MessageRouterPublishResponse.class);
+        when(response.successful()).thenReturn(isSuccess);
+        return response;
     }
 }
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapReRegistrationConsumerTaskImplTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapReRegistrationConsumerTaskImplTest.java
index 72e28987..edbe511d 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapReRegistrationConsumerTaskImplTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapReRegistrationConsumerTaskImplTest.java
@@ -20,38 +20,39 @@
 
 package org.onap.bbs.event.processor.tasks;
 
-import static org.mockito.Mockito.doReturn;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.reset;
+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 com.google.gson.Gson;
-import com.google.gson.JsonElement;
-
-import java.util.Optional;
-
-import javax.net.ssl.SSLException;
 
 import org.junit.Assert;
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.DmaapReRegistrationConsumerProperties;
 import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
 import org.onap.bbs.event.processor.model.ImmutableReRegistrationConsumerDmaapModel;
 import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
 import org.onap.bbs.event.processor.utilities.ReRegistrationDmaapConsumerJsonParser;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.DmaapConsumerConfiguration;
-import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.config.ImmutableDmaapConsumerConfiguration;
-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 org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.api.MessageRouterSubscriber;
+import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.model.config.MessageRouterSubscriberConfig;
 
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 class DmaapReRegistrationConsumerTaskImplTest {
 
+    private static final String DMAAP_PROTOCOL = "http";
+    private static final String DMAAP_HOST = "message-router.onap.svc.cluster.local";
+    private static final int DMAAP_PORT = 3904;
+    private static final String DMAAP_TOPIC = "unauthenticated.PNF_REREGISTRATION";
+    private static final String SUBSCRIBER_ID = "subscriberID";
+    private static final String SUBSCRIBER_GROUP = "subscriberGroup";
+
     private static final String RE_REGISTRATION_EVENT_TEMPLATE = "{\"event\": {"
             + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
             + "\"additionalFields\": {"
@@ -63,31 +64,33 @@ class DmaapReRegistrationConsumerTaskImplTest {
 
     private static DmaapReRegistrationConsumerTaskImpl dmaapConsumerTask;
     private static ReRegistrationConsumerDmaapModel reRegistrationConsumerDmaapModel;
-    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
     private static String eventsArray;
+    private static MessageRouterSubscriber subscriber;
     private static Gson gson = new Gson();
+    private static ApplicationConfiguration configuration;
 
     @BeforeAll
-    static void setUp() throws SSLException {
-
-        final String sourceName = "PNF-CorrelationId";
-        final String attachmentPoint = "olt2/2/2";
-        final String remoteId = "remoteId";
-        final String cvlan = "1005";
-        final String svlan = "100";
+    static void setUp() {
 
         // Mock Re-registration configuration
-        DmaapConsumerConfiguration dmaapConsumerConfiguration = testVersionOfDmaapConsumerConfiguration();
-        ApplicationConfiguration configuration = mock(ApplicationConfiguration.class);
-        when(configuration.getDmaapReRegistrationConsumerConfiguration()).thenReturn(dmaapConsumerConfiguration);
-
-        // Mock reactive DMaaP client
-        ConsumerReactiveHttpClientFactory httpClientFactory = mock(ConsumerReactiveHttpClientFactory.class);
-        dMaaPConsumerReactiveHttpClient = mock(DMaaPConsumerReactiveHttpClient.class);
-        doReturn(dMaaPConsumerReactiveHttpClient).when(httpClientFactory).create(dmaapConsumerConfiguration);
-
-        dmaapConsumerTask = new DmaapReRegistrationConsumerTaskImpl(configuration,
-                new ReRegistrationDmaapConsumerJsonParser(), httpClientFactory);
+        configuration = mock(ApplicationConfiguration.class);
+        var props = mock(DmaapReRegistrationConsumerProperties.class);
+        when(props.getDmaapProtocol()).thenReturn(DMAAP_PROTOCOL);
+        when(props.getDmaapHostName()).thenReturn(DMAAP_HOST);
+        when(props.getDmaapPortNumber()).thenReturn(DMAAP_PORT);
+        when(props.getDmaapTopicName()).thenReturn(DMAAP_TOPIC);
+        when(props.getConsumerId()).thenReturn(SUBSCRIBER_ID);
+        when(props.getConsumerGroup()).thenReturn(SUBSCRIBER_GROUP);
+        when(configuration.getDmaapReRegistrationConsumerProperties()).thenReturn(props);
+
+        var subscriberConfig = mock(MessageRouterSubscriberConfig.class);
+        when(configuration.getDmaapReRegistrationConsumerConfiguration()).thenReturn(subscriberConfig);
+
+        var sourceName = "PNF-CorrelationId";
+        var attachmentPoint = "olt2/2/2";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
         reRegistrationConsumerDmaapModel = ImmutableReRegistrationConsumerDmaapModel.builder()
                 .correlationId(sourceName)
@@ -97,58 +100,43 @@ class DmaapReRegistrationConsumerTaskImplTest {
                 .sVlan(svlan)
                 .build();
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE, sourceName, attachmentPoint, remoteId,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE, sourceName, attachmentPoint, remoteId,
                 cvlan, svlan);
 
         eventsArray = "[" + event + "]";
     }
 
-    @AfterEach
-    void resetMock() {
-        reset(dMaaPConsumerReactiveHttpClient);
-    }
-
     @Test
     void passingEmptyMessage_NothingHappens() {
-        JsonElement empty = gson.toJsonTree("");
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty())).thenReturn(Mono.just(empty));
+        var empty = gson.toJsonTree("");
+        subscriber = mock(MessageRouterSubscriber.class);
+        when(subscriber.getElements(any())).thenReturn(Flux.just(empty));
+
+        dmaapConsumerTask = new DmaapReRegistrationConsumerTaskImpl(configuration, subscriber,
+                new ReRegistrationDmaapConsumerJsonParser());
 
         StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
                 .expectSubscription()
                 .expectError(EmptyDmaapResponseException.class);
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
+
+        verify(subscriber, times(1)).getElements(any());
+        verifyNoMoreInteractions(subscriber);
     }
 
     @Test
     void passingNormalMessage_ResponseSucceeds() {
-        JsonElement normalEventsArray = gson.toJsonTree(eventsArray);
-        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse(Optional.empty()))
-                .thenReturn(Mono.just(normalEventsArray));
+        System.out.println("Events sent : " + eventsArray);
+        var normalEventsArray = gson.toJsonTree(eventsArray);
+        subscriber = mock(MessageRouterSubscriber.class);
+        when(subscriber.getElements(any())).thenReturn(Flux.just(normalEventsArray));
+
+        dmaapConsumerTask = new DmaapReRegistrationConsumerTaskImpl(configuration, subscriber,
+                new ReRegistrationDmaapConsumerJsonParser());
 
         StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
                 .expectSubscription()
                 .consumeNextWith(e -> Assert.assertEquals(e, reRegistrationConsumerDmaapModel));
-        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse(Optional.empty());
-    }
-
-    private static DmaapConsumerConfiguration testVersionOfDmaapConsumerConfiguration() {
-        return new ImmutableDmaapConsumerConfiguration.Builder()
-                .consumerGroup("OpenDCAE-c12")
-                .consumerId("c12")
-                .dmaapContentType("application/json")
-                .dmaapHostName("message-router.onap.svc.cluster.local")
-                .dmaapPortNumber(3904)
-                .dmaapProtocol("http")
-                .dmaapUserName("admin")
-                .dmaapUserPassword("admin")
-                .trustStorePath("/opt/app/bbs/local/org.onap.bbs.trust.jks")
-                .trustStorePasswordPath("change_it")
-                .keyStorePath("/opt/app/bbs/local/org.onap.bbs.p12")
-                .keyStorePasswordPath("change_it")
-                .enableDmaapCertAuth(false)
-                .dmaapTopicName("/events/unauthenticated.PNF_REREGISTRATION")
-                .timeoutMs(-1)
-                .messageLimit(-1)
-                .build();
+        verify(subscriber, times(1)).getElements(any());
+        verifyNoMoreInteractions(subscriber);
     }
 }
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/AaiReactiveClientTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/AaiReactiveClientTest.java
index 8e3c46ba..2876dd90 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/AaiReactiveClientTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/AaiReactiveClientTest.java
@@ -37,8 +37,6 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.ServiceLoader;
 
-import javax.net.ssl.SSLException;
-
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
@@ -74,13 +72,13 @@ class AaiReactiveClientTest {
     private static WireMockServer wireMockServer;
 
     @BeforeAll
-    static void init() throws SSLException {
-        GsonBuilder gsonBuilder = new GsonBuilder();
+    static void init() {
+        var gsonBuilder = new GsonBuilder();
         ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
         gson = gsonBuilder.create();
 
-        ApplicationConfiguration configuration = Mockito.mock(ApplicationConfiguration.class);
-        AaiClientConfiguration aaiClientConfiguration = Mockito.mock(AaiClientConfiguration.class);
+        var configuration = Mockito.mock(ApplicationConfiguration.class);
+        var aaiClientConfiguration = Mockito.mock(AaiClientConfiguration.class);
         when(configuration.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
         when(aaiClientConfiguration.aaiUserName()).thenReturn("AAI");
         when(aaiClientConfiguration.aaiUserPassword()).thenReturn("AAI");
@@ -106,10 +104,10 @@ class AaiReactiveClientTest {
     @Test
     void sendingReactiveRequestForPnf_Succeeds() {
 
-        String pnfName = "pnf-1";
-        String attachmentPoint = "olt1-1-1";
+        var pnfName = "pnf-1";
+        var attachmentPoint = "olt1-1-1";
 
-        String pnfUrl = String.format("/aai/v14/network/pnfs/pnf/%s?depth=1", pnfName);
+        var pnfUrl = String.format("/aai/v14/network/pnfs/pnf/%s?depth=1", pnfName);
 
         // Build Relationship Data
         RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
@@ -166,7 +164,7 @@ class AaiReactiveClientTest {
                 .expectSubscription()
                 .consumeNextWith(pnf -> {
                     Assertions.assertEquals(pnfName, pnf.getPnfName(), "PNF Name in response does not match");
-                    String extractedAttachmentPoint = pnf.getRelationshipListAaiObject().getRelationshipEntries()
+                    var extractedAttachmentPoint = pnf.getRelationshipListAaiObject().getRelationshipEntries()
                             .stream()
                             .filter(e -> e.getRelatedTo().equals("logical-link"))
                             .flatMap(e -> e.getRelationshipData().stream())
@@ -182,10 +180,10 @@ class AaiReactiveClientTest {
     @Test
     void sendingReactiveRequestForServiceInstance_Succeeds() {
 
-        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
-        String orchestrationStatus = "active";
+        var serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        var orchestrationStatus = "active";
 
-        String serviceInstanceUrl =
+        var serviceInstanceUrl =
                 String.format("/aai/v14/nodes/service-instances/service-instance/%s?format=resource_and_url",
                 serviceInstanceId);
 
@@ -237,10 +235,10 @@ class AaiReactiveClientTest {
                     Assertions.assertEquals(serviceInstanceId, serviceInstance.getServiceInstanceId(),
                             "Service Instance ID in response does not match");
 
-                    MetadataListAaiObject extractedMetadataListObject =
+                    var extractedMetadataListObject =
                             serviceInstance.getMetadataListAaiObject().orElseThrow(AaiReactiveClientTestException::new);
 
-                    MetadataListAaiObject.MetadataEntryAaiObject extractedMetadataEntry =
+                    var extractedMetadataEntry =
                             extractedMetadataListObject.getMetadataEntries()
                             .stream()
                             .filter(m -> m.getMetaname().equals("cvlan"))
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/CpeAuthenticationDmaapConsumerJsonParserTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/CpeAuthenticationDmaapConsumerJsonParserTest.java
index c7ad7935..d7970ae9 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/CpeAuthenticationDmaapConsumerJsonParserTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/CpeAuthenticationDmaapConsumerJsonParserTest.java
@@ -104,9 +104,8 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingNonJson_getIllegalStateException() {
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                new CpeAuthenticationDmaapConsumerJsonParser();
-        JsonReader jsonReader = new JsonReader(new StringReader("not JSON"));
+        var consumerJsonParser = new CpeAuthenticationDmaapConsumerJsonParser();
+        var jsonReader = new JsonReader(new StringReader("not JSON"));
         jsonReader.setLenient(true);
         JsonElement notJson = jsonParser.parse(jsonReader).getAsJsonPrimitive();
 
@@ -118,8 +117,7 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingNoEvents_EmptyFluxIsReturned() {
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                new CpeAuthenticationDmaapConsumerJsonParser();
+        var consumerJsonParser = new CpeAuthenticationDmaapConsumerJsonParser();
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse("[]"))))
                 .expectSubscription()
@@ -129,23 +127,22 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingOneCorrectEvent_validationSucceeds() {
 
-        String sourceName = "PNF-CorrelationId";
-        String oldAuthenticationState = "outOfService";
-        String newAuthenticationState = "inService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var sourceName = "PNF-CorrelationId";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName, oldAuthenticationState,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName, oldAuthenticationState,
                 newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         CpeAuthenticationConsumerDmaapModel expectedEventObject = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
                 .correlationId(sourceName)
@@ -164,30 +161,29 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingTwoCorrectEvents_validationSucceeds() {
 
-        String sourceName1 = "PNF-CorrelationId";
-        String sourceName2 = "PNF-CorrelationId";
-        String oldAuthenticationState = "outOfService";
-        String newAuthenticationState = "inService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress1 = "00:0a:95:8d:78:16";
-        String rgwMacAddress2 = "00:0a:95:8d:78:17";
-        String swVersion = "1.2";
+        var sourceName1 = "PNF-CorrelationId";
+        var sourceName2 = "PNF-CorrelationId";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress1 = "00:0a:95:8d:78:16";
+        var rgwMacAddress2 = "00:0a:95:8d:78:17";
+        var swVersion = "1.2";
 
-        String firstEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName1, oldAuthenticationState,
+        var firstEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName1, oldAuthenticationState,
                 newAuthenticationState, stateInterface, rgwMacAddress1, swVersion);
-        String secondEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName2, oldAuthenticationState,
+        var secondEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName2, oldAuthenticationState,
                 newAuthenticationState, stateInterface, rgwMacAddress2, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement1 = jsonParser.parse(firstEvent);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement1 = jsonParser.parse(firstEvent);
         Mockito.doReturn(Optional.of(jsonElement1.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement1);
-        JsonElement jsonElement2 = jsonParser.parse(secondEvent);
+        var jsonElement2 = jsonParser.parse(secondEvent);
         Mockito.doReturn(Optional.of(jsonElement2.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement2);
 
-        String eventsArray = "[" + firstEvent + "," + secondEvent + "]";
+        var eventsArray = "[" + firstEvent + "," + secondEvent + "]";
 
         CpeAuthenticationConsumerDmaapModel expectedFirstEventObject =
                 ImmutableCpeAuthenticationConsumerDmaapModel.builder()
@@ -217,22 +213,21 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingAuthenticationState_validationFails() {
 
-        String sourceName = "PNF-CorrelationId";
-        String oldAuthenticationState = "outOfService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var sourceName = "PNF-CorrelationId";
+        var oldAuthenticationState = "outOfService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_AUTHENTICATION_STATE, sourceName,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_AUTHENTICATION_STATE, sourceName,
                 oldAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -242,22 +237,21 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingSourceName_validationFails() {
 
-        String oldAuthenticationState = "outOfService";
-        String newAuthenticationState = "inService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME,
                 oldAuthenticationState, newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -267,22 +261,21 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingSourceNameValue_validationFails() {
 
-        String oldAuthenticationState = "outOfService";
-        String newAuthenticationState = "inService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME_VALUE,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME_VALUE,
                 oldAuthenticationState, newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -292,22 +285,21 @@ class CpeAuthenticationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingStateChangeFieldsHeader_validationFails() {
 
-        String oldAuthenticationState = "outOfService";
-        String newAuthenticationState = "inService";
-        String stateInterface = "stateInterface";
-        String rgwMacAddress = "00:0a:95:8d:78:16";
-        String swVersion = "1.2";
+        var oldAuthenticationState = "outOfService";
+        var newAuthenticationState = "inService";
+        var stateInterface = "stateInterface";
+        var rgwMacAddress = "00:0a:95:8d:78:16";
+        var swVersion = "1.2";
 
-        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_STATE_CHANGE_FIELDS,
+        var event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_STATE_CHANGE_FIELDS,
                 oldAuthenticationState, newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
 
-        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
-                spy(new CpeAuthenticationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/ReRegistrationDmaapConsumerJsonParserTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/ReRegistrationDmaapConsumerJsonParserTest.java
index cd238e2d..6d78826b 100644
--- a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/ReRegistrationDmaapConsumerJsonParserTest.java
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/ReRegistrationDmaapConsumerJsonParserTest.java
@@ -93,8 +93,8 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingNonJson_getIllegalStateException() {
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
-        JsonReader jsonReader = new JsonReader(new StringReader("not JSON"));
+        var consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
+        var jsonReader = new JsonReader(new StringReader("not JSON"));
         jsonReader.setLenient(true);
         JsonElement notJson = jsonParser.parse(jsonReader).getAsJsonPrimitive();
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(notJson)))
@@ -105,7 +105,7 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingNoEvents_EmptyFluxIsReturned() {
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
+        var consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse("[]"))))
                 .expectSubscription()
                 .verifyComplete();
@@ -114,21 +114,21 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingOneCorrectEvent_validationSucceeds() {
 
-        String correlationId = "PNF-CorrelationId";
-        String attachmentPoint = "olt1/1/1";
-        String remoteId = "remoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var correlationId = "PNF-CorrelationId";
+        var attachmentPoint = "olt1/1/1";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId, attachmentPoint,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId, attachmentPoint,
                 remoteId, cvlan, svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         ReRegistrationConsumerDmaapModel expectedEventObject = ImmutableReRegistrationConsumerDmaapModel.builder()
                 .correlationId(correlationId)
@@ -146,29 +146,29 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingTwoCorrectEvents_validationSucceeds() {
 
-        String correlationId1 = "PNF-CorrelationId1";
-        String correlationId2 = "PNF-CorrelationId2";
-        String attachmentPoint1 = "olt1/1/1";
-        String attachmentPoint2 = "olt2/2/2";
-        String remoteId1 = "remoteId1";
-        String remoteId2 = "remoteId2";
-        String cvlan = "1005";
-        String svlan = "100";
+        var correlationId1 = "PNF-CorrelationId1";
+        var correlationId2 = "PNF-CorrelationId2";
+        var attachmentPoint1 = "olt1/1/1";
+        var attachmentPoint2 = "olt2/2/2";
+        var remoteId1 = "remoteId1";
+        var remoteId2 = "remoteId2";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String firstEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
+        var firstEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
                 remoteId1, cvlan, svlan);
-        String secondEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
+        var secondEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
                 remoteId1, cvlan, svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement1 = jsonParser.parse(firstEvent);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement1 = jsonParser.parse(firstEvent);
         Mockito.doReturn(Optional.of(jsonElement1.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement1);
-        JsonElement jsonElement2 = jsonParser.parse(secondEvent);
+        var jsonElement2 = jsonParser.parse(secondEvent);
         Mockito.doReturn(Optional.of(jsonElement2.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement2);
 
-        String eventsArray = "[" + firstEvent + "," + secondEvent + "]";
+        var eventsArray = "[" + firstEvent + "," + secondEvent + "]";
 
         ReRegistrationConsumerDmaapModel expectedFirstEventObject = ImmutableReRegistrationConsumerDmaapModel.builder()
                 .correlationId(correlationId1)
@@ -194,23 +194,23 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingAttachmentPoint_validationFails() {
 
-        String correlationId = "PNF-CorrelationId";
-        String remoteId = "remoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var correlationId = "PNF-CorrelationId";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ATTACHMENT_POINT,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ATTACHMENT_POINT,
                 correlationId,
                 remoteId,
                 cvlan,
                 svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -220,23 +220,23 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingCorrelationId_validationFails() {
 
-        String attachmentPoint = "olt1/1/1";
-        String remoteId = "remoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var attachmentPoint = "olt1/1/1";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID,
                 attachmentPoint,
                 remoteId,
                 cvlan,
                 svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -246,23 +246,23 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingCorrelationIdValue_validationFails() {
 
-        String attachmentPoint = "olt1/1/1";
-        String remoteId = "remoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var attachmentPoint = "olt1/1/1";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID_VALUE,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID_VALUE,
                 attachmentPoint,
                 remoteId,
                 cvlan,
                 svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()
@@ -272,25 +272,25 @@ class ReRegistrationDmaapConsumerJsonParserTest {
     @Test
     void passingJsonWithMissingAdditionalFields_validationFails() {
 
-        String correlationId = "PNF-CorrelationId";
-        String attachmentPoint = "olt1/1/1";
-        String remoteId = "remoteId";
-        String cvlan = "1005";
-        String svlan = "100";
+        var correlationId = "PNF-CorrelationId";
+        var attachmentPoint = "olt1/1/1";
+        var remoteId = "remoteId";
+        var cvlan = "1005";
+        var svlan = "100";
 
-        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ADDITIONAL_FIELDS,
+        var event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ADDITIONAL_FIELDS,
                 correlationId,
                 attachmentPoint,
                 remoteId,
                 cvlan,
                 svlan);
 
-        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
-        JsonElement jsonElement = jsonParser.parse(event);
+        var consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        var jsonElement = jsonParser.parse(event);
         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
                 .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 
-        String eventsArray = "[" + event + "]";
+        var eventsArray = "[" + event + "]";
 
         StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(jsonParser.parse(eventsArray))))
                 .expectSubscription()