BBS use case event processor microservice code

BBS event processor code handling PNF-relocation internal events from PRH
and CPE authentication events raised by external Edge SDN M&C entities onto
ONAP platform

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

20 files changed, 4650 insertions, 0 deletions

diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/AaiClientPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/AaiClientPropertiesTest.java
new file mode 100644
index 00000000..449be724
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/AaiClientPropertiesTest.java
@@ -0,0 +1,70 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = AaiClientProperties.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.aai.client.aaiHost=test localhost",
+        "configs.aai.client.aaiPort=1234",
+        "configs.aai.client.aaiProtocol=https",
+        "configs.aai.client.aaiUserName=AAI",
+        "configs.aai.client.aaiUserPassword=AAI",
+        "configs.aai.client.aaiIgnoreSslCertificateErrors=true",
+        "configs.aai.client.aaiHeaders.X-FromAppId=bbs",
+        "configs.aai.client.aaiHeaders.X-TransactionId=9999",
+        "configs.aai.client.aaiHeaders.Accept=application/json",
+        "configs.aai.client.aaiHeaders.Real-Time=true",
+        "configs.aai.client.aaiHeaders.Content-Type=application/merge-patch+json"})
+class AaiClientPropertiesTest {
+
+    private AaiClientProperties properties;
+
+    @Autowired
+    public AaiClientPropertiesTest(
+            AaiClientProperties properties) {
+        this.properties = properties;
+    }
+
+    @Test
+    void dmaapReRegistrationProperties_SuccessFullyLoaded() {
+        assertEquals("test localhost", properties.getAaiHost());
+        assertEquals(1234, properties.getAaiPort());
+        assertEquals("https", properties.getAaiProtocol());
+        assertEquals("AAI", properties.getAaiUserName());
+        assertEquals("AAI", properties.getAaiUserPassword());
+        assertTrue(properties.isAaiIgnoreSslCertificateErrors());
+        assertEquals("bbs", properties.getAaiHeaders().get("X-FromAppId"));
+        assertEquals("9999", properties.getAaiHeaders().get("X-TransactionId"));
+        assertEquals("application/json", properties.getAaiHeaders().get("Accept"));
+        assertEquals("true", properties.getAaiHeaders().get("Real-Time"));
+        assertEquals("application/merge-patch+json", properties.getAaiHeaders().get("Content-Type"));
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..59feacaa
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ApplicationConfigurationTest.java
@@ -0,0 +1,350 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.FixMethodOrder;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.runners.MethodSorters;
+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;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = {
+        AaiClientProperties.class,
+        DmaapReRegistrationConsumerProperties.class,
+        DmaapCpeAuthenticationConsumerProperties.class,
+        DmaapProducerProperties.class,
+        SecurityProperties.class,
+        GenericProperties.class})
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.aai.client.aaiHost=test localhost",
+        "configs.aai.client.aaiPort=1234",
+        "configs.aai.client.aaiProtocol=https",
+        "configs.aai.client.aaiUserName=AAI",
+        "configs.aai.client.aaiUserPassword=AAI",
+        "configs.aai.client.aaiIgnoreSslCertificateErrors=true",
+        "configs.aai.client.aaiHeaders.X-FromAppId=bbs",
+        "configs.aai.client.aaiHeaders.X-TransactionId=9999",
+        "configs.aai.client.aaiHeaders.Accept=application/json",
+        "configs.aai.client.aaiHeaders.Real-Time=true",
+        "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.dmaapTopicName=/events/unauthenticated.PNF_REREGISTRATION",
+        "configs.dmaap.consumer.re-registration.dmaapProtocol=http",
+        "configs.dmaap.consumer.re-registration.dmaapContentType=application/json",
+        "configs.dmaap.consumer.re-registration.consumerId=c12",
+        "configs.dmaap.consumer.re-registration.consumerGroup=OpenDcae-c12",
+        "configs.dmaap.consumer.re-registration.timeoutMs=-1",
+        "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.dmaapTopicName=/events/unauthenticated.CPE_AUTHENTICATION",
+        "configs.dmaap.consumer.cpe-authentication.dmaapProtocol=http",
+        "configs.dmaap.consumer.cpe-authentication.dmaapContentType=application/json",
+        "configs.dmaap.consumer.cpe-authentication.consumerId=c12",
+        "configs.dmaap.consumer.cpe-authentication.consumerGroup=OpenDcae-c12",
+        "configs.dmaap.consumer.cpe-authentication.timeoutMs=-1",
+        "configs.dmaap.consumer.cpe-authentication.messageLimit=1",
+        "configs.dmaap.producer.dmaapHostName=test localhost",
+        "configs.dmaap.producer.dmaapPortNumber=1234",
+        "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.enableDmaapCertAuth=true",
+        "configs.security.enableAaiCertAuth=true",
+        "configs.application.pipelinesPollingIntervalSec=30",
+        "configs.application.pipelinesTimeoutSec=15",
+        "configs.application.re-registration.policyScope=reRegPolicyScope",
+        "configs.application.re-registration.clControlName=reRegControlName",
+        "configs.application.cpe-authentication.policyScope=cpeAuthPolicyScope",
+        "configs.application.cpe-authentication.clControlName=cpeAuthControlName"})
+@DisplayName("Application Configuration Unit-Tests")
+// Ordering tests because we need a first configuration population from @TestPropertySource
+// and then update of the config parameters based on a new Consul-retrieved Configuration object
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+class ApplicationConfigurationTest {
+
+    private ApplicationConfiguration configuration;
+
+    @Autowired
+    public ApplicationConfigurationTest(AaiClientProperties aaiClientProperties,
+                                        DmaapReRegistrationConsumerProperties dmaapReRegistrationConsumerProperties,
+                                        DmaapCpeAuthenticationConsumerProperties
+                                                    dmaapCpeAuthenticationConsumerProperties,
+                                        DmaapProducerProperties dmaapProducerProperties,
+                                        SecurityProperties securityProperties,
+                                        GenericProperties genericProperties) {
+        this.configuration = new ApplicationConfiguration(aaiClientProperties,
+                dmaapReRegistrationConsumerProperties,
+                dmaapCpeAuthenticationConsumerProperties,
+                dmaapProducerProperties,
+                securityProperties,
+                genericProperties);
+    }
+
+    @Test
+    void testA_configurationObjectSuccessfullyPopulated() {
+
+        AaiClientConfiguration aaiClientConfiguration = configuration.getAaiClientConfiguration();
+        assertAll("AAI Client Configuration Properties",
+            () -> assertEquals("test localhost", aaiClientConfiguration.aaiHost()),
+            () -> assertEquals(Integer.valueOf(1234), aaiClientConfiguration.aaiPort()),
+            () -> assertEquals("https", aaiClientConfiguration.aaiProtocol()),
+            () -> assertEquals("AAI", aaiClientConfiguration.aaiUserName()),
+            () -> assertEquals("AAI", aaiClientConfiguration.aaiUserPassword()),
+            () -> assertTrue(aaiClientConfiguration.aaiIgnoreSslCertificateErrors()),
+            () -> assertEquals("bbs", aaiClientConfiguration.aaiHeaders().get("X-FromAppId")),
+            () -> assertEquals("9999", aaiClientConfiguration.aaiHeaders().get("X-TransactionId")),
+            () -> assertEquals("application/json", aaiClientConfiguration.aaiHeaders().get("Accept")),
+            () -> assertEquals("true", aaiClientConfiguration.aaiHeaders().get("Real-Time")),
+            () -> assertEquals("application/merge-patch+json",
+                        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("/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())
+        );
+
+        DmaapConsumerConfiguration dmaapConsumerCpeAuthenticationConfig =
+                configuration.getDmaapCpeAuthenticationConsumerConfiguration();
+        assertAll("DMaaP Consumer CPE Authentication Configuration Properties",
+            () -> assertEquals("test localhost", dmaapConsumerCpeAuthenticationConfig.dmaapHostName()),
+            () -> assertEquals(Integer.valueOf(1234), dmaapConsumerCpeAuthenticationConfig.dmaapPortNumber()),
+            () -> 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())
+        );
+
+        DmaapPublisherConfiguration dmaapPublisherConfiguration = configuration.getDmaapPublisherConfiguration();
+        assertAll("DMaaP Publisher Configuration Properties",
+            () -> assertEquals("test localhost", dmaapPublisherConfiguration.dmaapHostName()),
+            () -> assertEquals(Integer.valueOf(1234), dmaapPublisherConfiguration.dmaapPortNumber()),
+            () -> assertEquals("/events/unauthenticated.DCAE_CL_OUTPUT",
+                    dmaapPublisherConfiguration.dmaapTopicName()),
+            () -> assertEquals("http", dmaapPublisherConfiguration.dmaapProtocol()),
+            () -> assertEquals("", dmaapPublisherConfiguration.dmaapUserName()),
+            () -> assertEquals("", dmaapPublisherConfiguration.dmaapUserPassword()),
+            () -> assertEquals("application/json", dmaapPublisherConfiguration.dmaapContentType())
+        );
+
+        assertAll("Generic Application Properties",
+            () -> assertEquals(30, configuration.getPipelinesPollingIntervalInSeconds()),
+            () -> assertEquals(15, configuration.getPipelinesTimeoutInSeconds()),
+            () -> assertEquals("reRegPolicyScope", configuration.getReRegistrationCloseLoopPolicyScope()),
+            () -> assertEquals("cpeAuthPolicyScope", configuration.getCpeAuthenticationCloseLoopPolicyScope()),
+            () -> assertEquals("reRegControlName", configuration.getReRegistrationCloseLoopControlName()),
+            () -> assertEquals("cpeAuthControlName", configuration.getCpeAuthenticationCloseLoopControlName())
+        );
+    }
+
+    @Test
+    void testB_passingNewConfiguration_UpdateSucceeds() {
+        Map<String, GeneratedAppConfigObject.StreamsObject> subscribes = new HashMap<>();
+
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo1 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router1.us:3901/events/unauthenticated.PNF_UPDATE")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject1 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo1)
+                .build();
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo2 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router2.us:3902/events/unauthenticated.CPE_AUTHENTICATION")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject2 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo2)
+                .build();
+
+        subscribes.put("config_key_1", streamsObject1);
+        subscribes.put("config_key_2", streamsObject2);
+
+        // Create Publishes Objects
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo3 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router3.us:3903/events/unauthenticated.DCAE_CL_OUTPUT")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject3 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo3)
+                .build();
+
+        // Final config object
+        GeneratedAppConfigObject updatedConfiguration = ImmutableGeneratedAppConfigObject.builder()
+                .dmaapProtocol("https")
+                .dmaapContentType("application/json")
+                .dmaapConsumerConsumerId("c13")
+                .dmaapConsumerConsumerGroup("OpenDcae-c13")
+                .dmaapMessageLimit(10)
+                .dmaapTimeoutMs(5)
+                .aaiHost("aai.onap.svc.cluster.local")
+                .aaiPort(8443)
+                .aaiProtocol("http")
+                .aaiUsername("AAI-update")
+                .aaiPassword("AAI-update")
+                .aaiIgnoreSslCertificateErrors(false)
+                .pipelinesPollingIntervalSec(20)
+                .pipelinesTimeoutSec(20)
+                .cbsPollingIntervalSec(180)
+                .reRegistrationPolicyScope("policyScope-update")
+                .reRegistrationClControlName("controlName-update")
+                .cpeAuthPolicyScope("policyScope-update")
+                .cpeAuthClControlName("controlName-update")
+                .reRegConfigKey("config_key_1")
+                .cpeAuthConfigKey("config_key_2")
+                .closeLoopConfigKey("config_key_3")
+                .streamSubscribesMap(subscribes)
+                .streamPublishesMap(Collections.singletonMap("config_key_3", streamsObject3))
+                .build();
+
+        // Update the configuration
+        configuration.updateCurrentConfiguration(updatedConfiguration);
+
+        AaiClientConfiguration aaiClientConfiguration = configuration.getAaiClientConfiguration();
+        assertAll("AAI Client Configuration Properties",
+            () -> assertEquals("aai.onap.svc.cluster.local", aaiClientConfiguration.aaiHost()),
+            () -> assertEquals(Integer.valueOf(8443), aaiClientConfiguration.aaiPort()),
+            () -> assertEquals("http", aaiClientConfiguration.aaiProtocol()),
+            () -> assertEquals("AAI-update", aaiClientConfiguration.aaiUserName()),
+            () -> assertEquals("AAI-update", aaiClientConfiguration.aaiUserPassword()),
+            () -> assertFalse(aaiClientConfiguration.aaiIgnoreSslCertificateErrors()),
+            () -> assertEquals("bbs", aaiClientConfiguration.aaiHeaders().get("X-FromAppId")),
+            () -> assertEquals("9999", aaiClientConfiguration.aaiHeaders().get("X-TransactionId")),
+            () -> assertEquals("application/json", aaiClientConfiguration.aaiHeaders().get("Accept")),
+            () -> assertEquals("true", aaiClientConfiguration.aaiHeaders().get("Real-Time")),
+            () -> assertEquals("application/merge-patch+json",
+                        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("", dmaapConsumerReRegistrationConfig.dmaapUserName()),
+            () -> assertEquals("", 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())
+        );
+
+        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("", dmaapConsumerCpeAuthenticationConfig.dmaapUserName()),
+            () -> assertEquals("", 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())
+        );
+
+        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("", dmaapPublisherConfiguration.dmaapUserName()),
+            () -> assertEquals("", dmaapPublisherConfiguration.dmaapUserPassword()),
+            () -> assertEquals("application/json", dmaapPublisherConfiguration.dmaapContentType())
+        );
+
+        assertAll("Generic Application Properties",
+            () -> assertEquals(20, configuration.getPipelinesPollingIntervalInSeconds()),
+            () -> assertEquals(20, configuration.getPipelinesTimeoutInSeconds()),
+            () -> assertEquals(180, configuration.getCbsPollingInterval()),
+            () -> assertEquals("policyScope-update", configuration.getReRegistrationCloseLoopPolicyScope()),
+            () -> assertEquals("policyScope-update", configuration.getCpeAuthenticationCloseLoopPolicyScope()),
+            () -> assertEquals("controlName-update", configuration.getReRegistrationCloseLoopControlName()),
+            () -> assertEquals("controlName-update", configuration.getCpeAuthenticationCloseLoopControlName())
+        );
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..ee75926b
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/ConsulConfigurationGatewayTest.java
@@ -0,0 +1,201 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.doReturn;
+
+import com.google.gson.JsonParser;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+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;
+
+class ConsulConfigurationGatewayTest {
+
+    private ConsulConfigurationGateway configurationGateway;
+    private static JsonParser jsonParser;
+
+    @BeforeAll
+    static void setup() {
+        jsonParser = new JsonParser();
+    }
+
+    ConsulConfigurationGatewayTest() {
+        ApplicationConfiguration configuration = Mockito.mock(ApplicationConfiguration.class);
+        this.configurationGateway = new ConsulConfigurationGateway(configuration);
+    }
+
+    @Test
+    void passingValidJson_constructsGeneratedAppConfigObject() {
+        final String validJson = "{"
+                + "\"dmaap.protocol\": \"http\","
+                + "\"dmaap.contentType\": \"application/json\","
+                + "\"dmaap.consumer.consumerId\": \"c12\","
+                + "\"dmaap.consumer.consumerGroup\": \"OpenDcae-c12\","
+                + "\"dmaap.messageLimit\": 1,"
+                + "\"dmaap.timeoutMs\": -1,"
+                + "\"aai.host\": \"aai.onap.svc.cluster.local\","
+                + "\"aai.port\": \"8443\","
+                + "\"aai.protocol\": \"https\","
+                + "\"aai.username\": \"AAI\","
+                + "\"aai.password\": \"AAI\","
+                + "\"aai.aaiIgnoreSslCertificateErrors\": true,"
+                + "\"application.pipelinesPollingIntervalSec\": 30,"
+                + "\"application.pipelinesTimeoutSec\": 15,"
+                + "\"application.cbsPollingIntervalSec\": 180,"
+                + "\"application.reregistration.policyScope\": \"policyScope\","
+                + "\"application.reregistration.clControlName\": \"controlName\","
+                + "\"application.cpe.authentication.policyScope\": \"policyScope\","
+                + "\"application.cpe.authentication.clControlName\": \"controlName\","
+                + "\"application.reregistration.configKey\": \"config_key_2\","
+                + "\"application.cpeAuth.configKey\": \"config_key_1\","
+                + "\"application.closeLoop.configKey\": \"config_key_3\","
+                + "\"streams_subscribes\": {"
+                + "\"config_key_1\": {"
+                + "\"type\": \"message_router\","
+                + "\"aaf_username\": \"some-user\","
+                + "\"aaf_password\": \"some-password\","
+                + "\"dmaap_info\": {"
+                + "\"client_id\": \"1500462518108\","
+                + "\"client_role\": \"com.dcae.member\","
+                + "\"location\": \"mtc00\","
+                + "\"topic_url\": \"https://we-are-message-router.us:3905/events/unauthenticated.CPE_AUTHENTICATION\""
+                + "}"
+                + "},"
+                + "\"config_key_2\": {"
+                + "\"type\": \"message_router\","
+                + "\"aaf_username\": \"some-user\","
+                + "\"aaf_password\": \"some-password\","
+                + "\"dmaap_info\": {"
+                + "\"client_id\": \"1500462518108\","
+                + "\"client_role\": \"com.dcae.member\","
+                + "\"location\": \"mtc00\","
+                + "\"topic_url\": \"https://we-are-message-router.us:3905/events/unauthenticated.PNF_UPDATE\""
+                + "}"
+                + "}"
+                + "},"
+                + "\"streams_publishes\": {"
+                + "\"config_key_3\": {"
+                + "\"type\": \"message_router\","
+                + "\"aaf_username\": \"some-user\","
+                + "\"aaf_password\": \"some-password\","
+                + "\"dmaap_info\": {"
+                + "\"client_id\": \"1500462518108\","
+                + "\"client_role\": \"com.dcae.member\","
+                + "\"location\": \"mtc00\","
+                + "\"topic_url\": \"https://we-are-message-router.us:3905/events/unauthenticated.DCAE_CL_OUTPUT\""
+                + "}"
+                + "}"
+                + "},"
+                + "\"services_calls\": {"
+                + "\"aai-interaction\": []"
+                + "}"
+                + "}";
+
+        // Create expected configuration
+        // Create Subscribes Objects
+        Map<String, GeneratedAppConfigObject.StreamsObject> subscribes = new HashMap<>();
+
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo1 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router.us:3905/events/unauthenticated.CPE_AUTHENTICATION")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject1 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo1)
+                .build();
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo2 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router.us:3905/events/unauthenticated.PNF_UPDATE")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject2 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo2)
+                .build();
+
+        subscribes.put("config_key_1", streamsObject1);
+        subscribes.put("config_key_2", streamsObject2);
+
+        // Create Publishes Objects
+        GeneratedAppConfigObject.DmaapInfo dmaapInfo3 = ImmutableDmaapInfo.builder()
+                .clientId("1500462518108")
+                .clientRole("com.dcae.member")
+                .location("mtc00")
+                .topicUrl("https://we-are-message-router.us:3905/events/unauthenticated.DCAE_CL_OUTPUT")
+                .build();
+        GeneratedAppConfigObject.StreamsObject streamsObject3 = ImmutableStreamsObject.builder()
+                .type("message_router")
+                .aafUsername("some-user")
+                .aafPassword("some-password")
+                .dmaapInfo(dmaapInfo3)
+                .build();
+
+        // Expected final config object
+        GeneratedAppConfigObject expectedConfiguration = ImmutableGeneratedAppConfigObject.builder()
+                .dmaapProtocol("http")
+                .dmaapContentType("application/json")
+                .dmaapConsumerConsumerId("c12")
+                .dmaapConsumerConsumerGroup("OpenDcae-c12")
+                .dmaapMessageLimit(1)
+                .dmaapTimeoutMs(-1)
+                .aaiHost("aai.onap.svc.cluster.local")
+                .aaiPort(8443)
+                .aaiProtocol("https")
+                .aaiUsername("AAI")
+                .aaiPassword("AAI")
+                .aaiIgnoreSslCertificateErrors(true)
+                .pipelinesPollingIntervalSec(30)
+                .pipelinesTimeoutSec(15)
+                .cbsPollingIntervalSec(180)
+                .reRegistrationPolicyScope("policyScope")
+                .reRegistrationClControlName("controlName")
+                .cpeAuthPolicyScope("policyScope")
+                .cpeAuthClControlName("controlName")
+                .reRegConfigKey("config_key_2")
+                .cpeAuthConfigKey("config_key_1")
+                .closeLoopConfigKey("config_key_3")
+                .streamSubscribesMap(subscribes)
+                .streamPublishesMap(Collections.singletonMap("config_key_3", streamsObject3))
+                .build();
+
+        ConsulConfigurationGateway spiedGateway = Mockito.spy(configurationGateway);
+        doReturn(false).when(spiedGateway).environmentNotReady();
+        assertEquals(expectedConfiguration,
+                spiedGateway.generateAppConfigObject(jsonParser.parse(validJson).getAsJsonObject()));
+    }
+}
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapCpeAuthenticationConsumerPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapCpeAuthenticationConsumerPropertiesTest.java
new file mode 100644
index 00000000..103335ce
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapCpeAuthenticationConsumerPropertiesTest.java
@@ -0,0 +1,68 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = DmaapCpeAuthenticationConsumerProperties.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.dmaap.consumer.cpe-authentication.dmaapHostName=test localhost",
+        "configs.dmaap.consumer.cpe-authentication.dmaapPortNumber=1234",
+        "configs.dmaap.consumer.cpe-authentication.dmaapTopicName=/events/unauthenticated.PNF_CPE_AUTHENTICATION",
+        "configs.dmaap.consumer.cpe-authentication.dmaapProtocol=http",
+        "configs.dmaap.consumer.cpe-authentication.dmaapContentType=application/json",
+        "configs.dmaap.consumer.cpe-authentication.consumerId=c12",
+        "configs.dmaap.consumer.cpe-authentication.consumerGroup=OpenDcae-c12",
+        "configs.dmaap.consumer.cpe-authentication.timeoutMs=-1",
+        "configs.dmaap.consumer.cpe-authentication.messageLimit=1"})
+class DmaapCpeAuthenticationConsumerPropertiesTest {
+
+    private DmaapCpeAuthenticationConsumerProperties properties;
+
+    @Autowired
+    public DmaapCpeAuthenticationConsumerPropertiesTest(
+            DmaapCpeAuthenticationConsumerProperties properties) {
+        this.properties = properties;
+    }
+
+    @Test
+    void dmaapCpeAuthenticationProperties_SuccessFullyLoaded() {
+        assertEquals("test localhost", properties.getDmaapHostName());
+        assertEquals(1234, properties.getDmaapPortNumber());
+        assertEquals("/events/unauthenticated.PNF_CPE_AUTHENTICATION", properties.getDmaapTopicName());
+        assertEquals("http", properties.getDmaapProtocol());
+        assertNull(properties.getDmaapUserName());
+        assertNull(properties.getDmaapUserPassword());
+        assertEquals("application/json", properties.getDmaapContentType());
+        assertEquals("c12", properties.getConsumerId());
+        assertEquals("OpenDcae-c12", properties.getConsumerGroup());
+        assertEquals(-1, properties.getTimeoutMs());
+        assertEquals(1, properties.getMessageLimit());
+    }
+}
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapProducerPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapProducerPropertiesTest.java
new file mode 100644
index 00000000..7ba38f9e
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapProducerPropertiesTest.java
@@ -0,0 +1,60 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = DmaapProducerProperties.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.dmaap.producer.dmaapHostName=test localhost",
+        "configs.dmaap.producer.dmaapPortNumber=1234",
+        "configs.dmaap.producer.dmaapTopicName=/events/unauthenticated.DCAE_CL_OUTPUT",
+        "configs.dmaap.producer.dmaapProtocol=http",
+        "configs.dmaap.producer.dmaapContentType=application/json"})
+class DmaapProducerPropertiesTest {
+
+    private DmaapProducerProperties properties;
+
+    @Autowired
+    public DmaapProducerPropertiesTest(
+            DmaapProducerProperties properties) {
+        this.properties = properties;
+    }
+
+    @Test
+    void dmaapReRegistrationProperties_SuccessFullyLoaded() {
+        assertEquals("test localhost", properties.getDmaapHostName());
+        assertEquals(1234, properties.getDmaapPortNumber());
+        assertEquals("/events/unauthenticated.DCAE_CL_OUTPUT", properties.getDmaapTopicName());
+        assertEquals("http", properties.getDmaapProtocol());
+        assertNull(properties.getDmaapUserName());
+        assertNull(properties.getDmaapUserPassword());
+        assertEquals("application/json", properties.getDmaapContentType());
+    }
+}
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapReRegistrationConsumerPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapReRegistrationConsumerPropertiesTest.java
new file mode 100644
index 00000000..682d2f31
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/DmaapReRegistrationConsumerPropertiesTest.java
@@ -0,0 +1,68 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = DmaapReRegistrationConsumerProperties.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.dmaap.consumer.re-registration.dmaapHostName=test localhost",
+        "configs.dmaap.consumer.re-registration.dmaapPortNumber=1234",
+        "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",
+        "configs.dmaap.consumer.re-registration.consumerId=c12",
+        "configs.dmaap.consumer.re-registration.consumerGroup=OpenDcae-c12",
+        "configs.dmaap.consumer.re-registration.timeoutMs=-1",
+        "configs.dmaap.consumer.re-registration.messageLimit=1"})
+class DmaapReRegistrationConsumerPropertiesTest {
+
+    private DmaapReRegistrationConsumerProperties properties;
+
+    @Autowired
+    public DmaapReRegistrationConsumerPropertiesTest(
+            DmaapReRegistrationConsumerProperties properties) {
+        this.properties = properties;
+    }
+
+    @Test
+    void dmaapReRegistrationProperties_SuccessFullyLoaded() {
+        assertEquals("test localhost", properties.getDmaapHostName());
+        assertEquals(1234, properties.getDmaapPortNumber());
+        assertEquals("/events/unauthenticated.PNF_REREGISTRATION", properties.getDmaapTopicName());
+        assertEquals("http", properties.getDmaapProtocol());
+        assertNull(properties.getDmaapUserName());
+        assertNull(properties.getDmaapUserPassword());
+        assertEquals("application/json", properties.getDmaapContentType());
+        assertEquals("c12", properties.getConsumerId());
+        assertEquals("OpenDcae-c12", properties.getConsumerGroup());
+        assertEquals(-1, properties.getTimeoutMs());
+        assertEquals(1, properties.getMessageLimit());
+    }
+}
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/GenericPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/GenericPropertiesTest.java
new file mode 100644
index 00000000..12e393a1
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/GenericPropertiesTest.java
@@ -0,0 +1,65 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = {GenericProperties.class,
+        GenericProperties.ReRegistrationGenericProperties.class,
+        GenericProperties.CpeAuthenticationGenericProperties.class})
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "configs.application.pipelinesPollingIntervalSec=30",
+        "configs.application.pipelinesTimeoutSec=15",
+        "configs.application.re-registration.policyScope=reRegPolicyScope",
+        "configs.application.re-registration.clControlName=reRegControlName",
+        "configs.application.cpe-authentication.policyScope=cpeAuthPolicyScope",
+        "configs.application.cpe-authentication.clControlName=cpeAuthControlName"})
+class GenericPropertiesTest {
+
+    private GenericProperties genericProperties;
+
+    @Autowired
+    public GenericPropertiesTest(GenericProperties genericProperties) {
+        this.genericProperties = genericProperties;
+    }
+
+    @Test
+    void genericProperties_SuccessFullyLoaded() {
+        assertAll("Generic Application Properties",
+            () -> assertEquals(30, genericProperties.getPipelinesPollingIntervalSec()),
+            () -> assertEquals(15, genericProperties.getPipelinesTimeoutSec()),
+            () -> assertEquals("reRegPolicyScope", genericProperties.getReRegistration().getPolicyScope()),
+            () -> assertEquals("cpeAuthPolicyScope",
+                    genericProperties.getCpeAuthentication().getPolicyScope()),
+            () -> assertEquals("reRegControlName", genericProperties.getReRegistration().getClControlName()),
+            () -> assertEquals("cpeAuthControlName",
+                    genericProperties.getCpeAuthentication().getClControlName())
+        );
+    }
+}
\ No newline at end of file
diff --git a/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/SecurityPropertiesTest.java b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/SecurityPropertiesTest.java
new file mode 100644
index 00000000..0dd6d75d
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/config/SecurityPropertiesTest.java
@@ -0,0 +1,63 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.TestPropertySource;
+
+@SpringBootTest(classes = SecurityProperties.class)
+@EnableConfigurationProperties
+@TestPropertySource(properties = {
+        "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.enableDmaapCertAuth=true",
+        "configs.security.enableAaiCertAuth=true"})
+class SecurityPropertiesTest {
+
+    private SecurityProperties securityProperties;
+
+    @Autowired
+    public SecurityPropertiesTest(SecurityProperties securityProperties) {
+        this.securityProperties = securityProperties;
+    }
+
+    @Test
+    void securityProperties_SuccessFullyLoaded() {
+        assertEquals("test key store path",
+                securityProperties.getKeyStorePath());
+        assertEquals("test key store password path",
+                securityProperties.getKeyStorePasswordPath());
+        assertEquals("test trust store path",
+                securityProperties.getTrustStorePath());
+        assertEquals("test trust store password path",
+                securityProperties.getTrustStorePasswordPath());
+        assertTrue(securityProperties.isEnableDmaapCertAuth());
+        assertTrue(securityProperties.isEnableAaiCertAuth());
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..bacb6c3e
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/controllers/BbsEventProcessorControllerTest.java
@@ -0,0 +1,163 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.controllers;
+
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.pipelines.CpeAuthenticationPipeline;
+import org.onap.bbs.event.processor.pipelines.ReRegistrationPipeline;
+import org.onap.bbs.event.processor.pipelines.Scheduler;
+import org.springframework.beans.factory.annotation.Autowired;
+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")
+class BbsEventProcessorControllerTest {
+
+    @Autowired
+    private MockMvc mockMvc;
+
+    @Autowired
+    private ReRegistrationPipeline reRegistrationPipeline;
+    @Autowired
+    private CpeAuthenticationPipeline cpeAuthenticationPipeline;
+    @Autowired
+    private Scheduler scheduler;
+    @Autowired
+    private ApplicationConfiguration configuration;
+
+    @BeforeEach
+    void resetInteractions() {
+        Mockito.reset(scheduler);
+    }
+
+    @Test
+    void sendingHeartBeatRestCall_RespondsWithAlive() throws Exception {
+        MvcResult heartBeatResult = mockMvc.perform(get("/heartbeat")).andReturn();
+
+        mockMvc.perform(asyncDispatch(heartBeatResult))
+                .andExpect(status().isOk())
+                .andExpect(content().string("bbs-event-processor is alive\n"));
+    }
+
+    @Test
+    void sendingReRegistrationSubmissionRestCall_RespondsWithOk() throws Exception {
+        MvcResult reregistrationSubmissionResult = mockMvc.perform(post("/poll-reregistration-events")).andReturn();
+
+        mockMvc.perform(asyncDispatch(reregistrationSubmissionResult))
+                .andExpect(status().isOk())
+                .andExpect(content().string("Request submitted\n"));
+        verify(reRegistrationPipeline, timeout(500)).processPnfReRegistrationEvents();
+    }
+
+    @Test
+    void sendingCpeAuthenticationSubmissionRestCall_RespondsWithOk() throws Exception {
+        MvcResult reregistrationSubmissionResult = mockMvc.perform(post("/poll-cpe-authentication-events")).andReturn();
+
+        mockMvc.perform(asyncDispatch(reregistrationSubmissionResult))
+                .andExpect(status().isOk())
+                .andExpect(content().string("Request submitted\n"));
+        verify(cpeAuthenticationPipeline, timeout(500)).processPnfCpeAuthenticationEvents();
+    }
+
+    @Test
+    void sendingStartTasksRestCall_ifItSucceeds_RespondsWithOk() throws Exception {
+        when(scheduler.reScheduleProcessingTasks()).thenReturn(true);
+        MvcResult startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
+
+        mockMvc.perform(asyncDispatch(startTasksResult))
+                .andExpect(status().isOk())
+                .andExpect(content().string("Initiation of tasks was successful\n"));
+        verify(scheduler).reScheduleProcessingTasks();
+    }
+
+    @Test
+    void sendingStartTasksRestCall_ifItFails_RespondsWithNotAcceptable() throws Exception {
+        when(scheduler.reScheduleProcessingTasks()).thenReturn(false);
+        MvcResult startTasksResult = mockMvc.perform(post("/start-tasks")).andReturn();
+
+        mockMvc.perform(asyncDispatch(startTasksResult))
+                .andExpect(status().isNotAcceptable())
+                .andExpect(content().string("Initiation of tasks failed\n"));
+        verify(scheduler).reScheduleProcessingTasks();
+    }
+
+    @Test
+    void sendingCancelTasksRestCall_ifItSucceeds_RespondsWithOk() throws Exception {
+        when(scheduler.cancelScheduledProcessingTasks()).thenReturn(true);
+        MvcResult cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
+
+        mockMvc.perform(asyncDispatch(cancellationResult))
+                .andExpect(status().isOk())
+                .andExpect(content().string("Cancellation was successful\n"));
+        verify(scheduler).cancelScheduledProcessingTasks();
+    }
+
+    @Test
+    void sendingCancelTasksRestCall_ifItFails_RespondsWithNotAcceptable() throws Exception {
+        when(scheduler.cancelScheduledProcessingTasks()).thenReturn(false);
+        MvcResult cancellationResult = mockMvc.perform(post("/cancel-tasks")).andReturn();
+
+        mockMvc.perform(asyncDispatch(cancellationResult))
+                .andExpect(status().isNotAcceptable())
+                .andExpect(content().string("Cancellation failed\n"));
+        verify(scheduler).cancelScheduledProcessingTasks();
+    }
+
+    @TestConfiguration
+    static class ControllerTestConfiguration {
+        @Bean
+        ReRegistrationPipeline reRegistrationPipeline() {
+            return Mockito.mock(ReRegistrationPipeline.class);
+        }
+
+        @Bean
+        CpeAuthenticationPipeline cpeAuthenticationPipeline() {
+            return Mockito.mock(CpeAuthenticationPipeline.class);
+        }
+
+        @Bean
+        Scheduler scheduler() {
+            return Mockito.mock(Scheduler.class);
+        }
+
+        @Bean
+        ApplicationConfiguration configuration() {
+            return Mockito.mock(ApplicationConfiguration.class);
+        }
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..fd43b8be
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/model/GsonSerializationsTest.java
@@ -0,0 +1,379 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.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;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.ServiceLoader;
+
+import org.junit.jupiter.api.Test;
+import org.onap.bbs.event.processor.utilities.ControlLoopJsonBodyBuilder;
+import org.onap.bbs.event.processor.utilities.CpeAuthenticationJsonBodyBuilder;
+import org.onap.bbs.event.processor.utilities.ReRegistrationJsonBodyBuilder;
+
+class GsonSerializationsTest {
+
+    @Test
+    void creatingReRegistrationJsonBody_returnsJsonInString() {
+
+        String correlationId = "NokiaCorrelationId";
+        String attachmentPoint = "olt2/1/1";
+        String remoteId = "RemoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String template = "{"
+                + "\"correlationId\":\"%s\","
+                + "\"attachment-point\":\"%s\","
+                + "\"remote-id\":\"%s\","
+                + "\"cvlan\":\"%s\","
+                + "\"svlan\":\"%s\""
+                + "}";
+
+        ReRegistrationConsumerDmaapModel model = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(correlationId)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+
+        String expectedResult = String.format(template, correlationId, attachmentPoint, remoteId, cvlan, svlan);
+
+        assertEquals(expectedResult, new ReRegistrationJsonBodyBuilder().createJsonBody(model));
+    }
+
+    @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";
+
+        String template = "{"
+                + "\"correlationId\":\"%s\","
+                + "\"old-authentication-state\":\"%s\","
+                + "\"new-authentication-state\":\"%s\","
+                + "\"state-interface\":\"%s\","
+                + "\"rgw-mac-address\":\"%s\","
+                + "\"sw-version\":\"%s\""
+                + "}";
+
+        CpeAuthenticationConsumerDmaapModel model = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(correlationId)
+                .oldAuthenticationState(oldAuthenticationState.getNameInOnap())
+                .newAuthenticationState(newAuthenticationState.getNameInOnap())
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+
+        String expectedResult = String.format(template, correlationId, oldAuthenticationState.getNameInOnap(),
+                newAuthenticationState.getNameInOnap(), stateInterface, rgwMacAddress, swVersion);
+
+        assertEquals(expectedResult, new CpeAuthenticationJsonBodyBuilder().createJsonBody(model));
+    }
+
+    @Test
+    void creatingDcaeControlLoopJsonBody_returnsJsonInString() {
+
+        String closedLoopEventClient = "DCAE.BBS_mSInstance";
+        String policyVersion = "1.0.0.5";
+        String policyName = "CPE_Authentication";
+        String 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";
+
+        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 = "{"
+                + "\"closedLoopEventClient\":\"%s\","
+                + "\"policyVersion\":\"%s\","
+                + "\"policyName\":\"%s\","
+                + "\"policyScope\":\"%s\","
+                + "\"target_type\":\"%s\","
+                + "\"AAI\":{"
+                + "\"service-information.service-instance-id\":\"service-instance-id-example\","
+                + "\"cvlan-id\":\"example cvlan-id\","
+                + "\"svlan-id\":\"example svlan-id\""
+                + "},"
+                + "\"closedLoopAlarmStart\":%s,"
+                + "\"closedLoopEventStatus\":\"%s\","
+                + "\"closedLoopControlName\":\"%s\","
+                + "\"version\":\"%s\","
+                + "\"target\":\"%s\","
+                + "\"requestID\":\"%s\","
+                + "\"from\":\"%s\""
+                + "}";
+
+
+        ControlLoopPublisherDmaapModel model = ImmutableControlLoopPublisherDmaapModel.builder()
+                .closedLoopEventClient(closedLoopEventClient)
+                .policyVersion(policyVersion)
+                .policyName(policyName)
+                .policyScope(policyScope)
+                .targetType(targetType)
+                .aaiEnrichmentData(aaiEnrichmentData)
+                .closedLoopAlarmStart(closedLoopAlarmStart)
+                .closedLoopEventStatus(closedLoopEventStatus)
+                .closedLoopControlName(closedLoopControlName)
+                .version(version)
+                .target(target)
+                .requestId(requestId)
+                .originator(from)
+                .build();
+
+        String expectedResult = String.format(template,
+                closedLoopEventClient,
+                policyVersion,
+                policyName,
+                policyScope,
+                targetType,
+                closedLoopAlarmStart,
+                closedLoopEventStatus,
+                closedLoopControlName,
+                version,
+                target,
+                requestId,
+                from);
+
+        assertEquals(expectedResult, new ControlLoopJsonBodyBuilder().createJsonBody(model));
+    }
+
+    @Test
+    void pnfAaiObject_IsSerializedSuccessfully() {
+
+        GsonBuilder gsonBuilder = new GsonBuilder();
+        ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
+        Gson gson = gsonBuilder.create();
+
+        String pnfName = "NokiaCorrelationID";
+        String swVersion = "1.2";
+
+        String template = "{"
+                + "\"pnf-name\":\"%s\","
+                + "\"in-maint\":true,"
+                + "\"sw-version\":\"%s\","
+                + "\"relationship-list\":{"
+                + "\"relationship\":["
+                + "{"
+                + "\"related-to\":\"service-instance\","
+                + "\"relationship-label\":\"org.onap.relationships.inventory.ComposedOf\","
+                + "\"related-link\":\"/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                + "/service-subscription/BBS/service-instances/service-instance/84003b26-6b76-4c75-b805-7b14ab4ffaef\","
+                + "\"relationship-data\":["
+                + "{"
+                + "\"relationship-key\":\"customer.global-customer-id\","
+                + "\"relationship-value\":\"Demonstration\""
+                + "},"
+                + "{"
+                + "\"relationship-key\":\"service-subscription.service-type\","
+                + "\"relationship-value\":\"BBS\""
+                + "},"
+                + "{"
+                + "\"relationship-key\":\"service-instance.service-instance-id\","
+                + "\"relationship-value\":\"84003b26-6b76-4c75-b805-7b14ab4ffaef\""
+                + "}"
+                + "],"
+                + "\"related-to-property\":["
+                + "{"
+                + "\"property-key\":\"service-instance.service-instance-name\","
+                + "\"property-value\":\"bbs-instance\""
+                + "}"
+                + "]"
+                + "},"
+                + "{"
+                + "\"related-to\":\"platform\","
+                + "\"relationship-label\":\"org.onap.relationships.inventory.Uses\","
+                + "\"related-link\":\"/aai/v14/business/platforms/platform/bbs-platform\","
+                + "\"relationship-data\":["
+                + "{"
+                + "\"relationship-key\":\"platform.platform-name\","
+                + "\"relationship-value\":\"bbs-platform\""
+                + "}"
+                + "]"
+                + "}"
+                + "]"
+                + "}"
+                + "}";
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                            + "/service-subscription/BBS/service-instances"
+                            + "/service-instance/84003b26-6b76-4c75-b805-7b14ab4ffaef")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                            ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("customer.global-customer-id")
+                                .relationshipValue("Demonstration").build(),
+                            ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("service-subscription.service-type")
+                                .relationshipValue("BBS").build(),
+                            ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("service-instance.service-instance-id")
+                                .relationshipValue("84003b26-6b76-4c75-b805-7b14ab4ffaef").build())
+                )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                .build();
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject secondRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                .relatedTo("platform")
+                .relatedLink("/aai/v14/business/platforms/platform/bbs-platform")
+                .relationshipLabel("org.onap.relationships.inventory.Uses")
+                .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                        .relationshipKey("platform.platform-name")
+                        .relationshipValue("bbs-platform").build()))
+                .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Arrays.asList(firstRelationshipEntry, secondRelationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        PnfAaiObject pnfAaiObject = ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .swVersion(swVersion)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+
+
+        String jsonPnfObject = String.format(template, pnfName, swVersion);
+
+        assertEquals(jsonPnfObject, gson.toJson(pnfAaiObject));
+        assertEquals(pnfAaiObject, gson.fromJson(jsonPnfObject, ImmutablePnfAaiObject.class));
+    }
+
+    @Test
+    void serviceInstanceAaiObject_IsSerializedSuccessfully() {
+
+        GsonBuilder gsonBuilder = new GsonBuilder();
+        ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
+        Gson gson = gsonBuilder.create();
+
+        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        String orchestrationStatus = "active";
+
+        String template = "{"
+                + "\"service-instance-id\":\"%s\","
+                + "\"orchestration-status\":\"%s\","
+                + "\"relationship-list\":{"
+                + "\"relationship\":["
+                + "{"
+                + "\"related-to\":\"service-instance\","
+                + "\"relationship-label\":\"org.onap.relationships.inventory.ComposedOf\","
+                + "\"related-link\":\"/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                + "/service-subscription/BBS-CFS"
+                + "/service-instances/service-instance/bb374844-44e4-488f-8381-fb5a0e3e6989\","
+                + "\"relationship-data\":["
+                + "{"
+                + "\"relationship-key\":\"service-instance.service-instance-id\","
+                + "\"relationship-value\":\"bb374844-44e4-488f-8381-fb5a0e3e6989\""
+                + "}"
+                + "]"
+                + "}"
+                + "]"
+                + "},"
+                + "\"metadata\":{"
+                + "\"metadatum\":["
+                + "{"
+                + "\"metaname\":\"cvlan\","
+                + "\"metaval\":\"1005\""
+                + "}"
+                + "]"
+                + "}"
+                + "}";
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/bb374844-44e4-488f-8381-fb5a0e3e6989")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("service-instance.service-instance-id")
+                                .relationshipValue("bb374844-44e4-488f-8381-fb5a0e3e6989").build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
+                ImmutableMetadataEntryAaiObject.builder()
+                .metaname("cvlan")
+                .metavalue("1005")
+                .build();
+
+        MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
+                .metadataEntries(Collections.singletonList(metadataEntry))
+                .build();
+
+        // Finally construct Service Instance object data
+        ServiceInstanceAaiObject serviceInstanceAaiObject = ImmutableServiceInstanceAaiObject.builder()
+                .serviceInstanceId(serviceInstanceId)
+                .orchestrationStatus(orchestrationStatus)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .metadataListAaiObject(metadataListAaiObject)
+                .build();
+
+
+        String jsonServiceInstanceObject = String.format(template, serviceInstanceId, orchestrationStatus);
+
+        assertEquals(jsonServiceInstanceObject, gson.toJson(serviceInstanceAaiObject));
+        assertEquals(serviceInstanceAaiObject, gson.fromJson(jsonServiceInstanceObject,
+                ImmutableServiceInstanceAaiObject.class));
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..dbac5bf2
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/CpeAuthenticationPipelineTest.java
@@ -0,0 +1,656 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.pipelines;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_CPE_AUTHENTICATION_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.UUID;
+
+import javax.net.ssl.SSLException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.exceptions.AaiTaskException;
+import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
+import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
+import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.ImmutableCpeAuthenticationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.ImmutableMetadataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableMetadataListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePnfAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePropertyAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipDataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableServiceInstanceAaiObject;
+import org.onap.bbs.event.processor.model.MetadataListAaiObject;
+import org.onap.bbs.event.processor.model.PnfAaiObject;
+import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
+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.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+// We can safely suppress unchecked assignment warnings for the ResponseEntity mock
+@SuppressWarnings("unchecked")
+@DisplayName("CPE Authentication Pipeline Unit-Tests")
+class CpeAuthenticationPipelineTest {
+
+    private CpeAuthenticationPipeline pipeline;
+    private ApplicationConfiguration configuration;
+    private DmaapCpeAuthenticationConsumerTask consumerTask;
+    private DmaapPublisherTask publisherTask;
+    private AaiClientTask aaiClientTask;
+
+    private ResponseEntity<String> responseEntity;
+
+    @BeforeEach
+    void setup() {
+
+        responseEntity = Mockito.mock(ResponseEntity.class);
+
+        configuration = Mockito.mock(ApplicationConfiguration.class);
+        consumerTask = Mockito.mock(DmaapCpeAuthenticationConsumerTask.class);
+        publisherTask = Mockito.mock(DmaapPublisherTask.class);
+        aaiClientTask = Mockito.mock(AaiClientTask.class);
+
+        when(configuration.getCpeAuthenticationCloseLoopControlName())
+                .thenReturn("controlName");
+        when(configuration.getCpeAuthenticationCloseLoopPolicyScope())
+                .thenReturn("policyScope");
+
+        pipeline = new CpeAuthenticationPipeline(configuration, consumerTask,
+                publisherTask, aaiClientTask, new HashMap<>());
+    }
+
+    @Test
+    void handleEmptyResponseFromDmaap() throws SSLException {
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(anyString()))
+                .thenReturn(Flux.error(new EmptyDmaapResponseException("Mock empty")));
+
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @Test
+    void noResponseFromDmaap_PipelineTimesOut() throws SSLException {
+
+        // Prepare mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME))
+                .thenReturn(Flux.never());
+
+        // Execute pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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";
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        // Prepare mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME)).thenReturn(Flux.just(event));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString())).thenReturn(Mono.never());
+
+        // Execute pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, rgwMacAddress);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME)).thenReturn(Flux.just(event));
+
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject));
+
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.never());
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, rgwMacAddress);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME)).thenReturn(Flux.just(event));
+
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject));
+
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // 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, 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",
+                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(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(publisherTask, times(2)).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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";
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel event = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        // Prepare Mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME)).thenReturn(Flux.just(event));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verify(aaiClientTask).executePnfRetrieval(anyString(), anyString());
+        verifyNoMoreInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .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();
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+        CpeAuthenticationConsumerDmaapModel secondEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName1, hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, rgwMacAddress);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME))
+                .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(aaiClientTask, times(2)).executePnfRetrieval(anyString(), anyString());
+        verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        CpeAuthenticationConsumerDmaapModel firstEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+        CpeAuthenticationConsumerDmaapModel secondEvent = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName2, hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, rgwMacAddress);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_CPE_AUTHENTICATION_TASK_NAME))
+                .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")))
+                .thenReturn(Mono.just(pnfAaiObject));
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(aaiClientTask, times(2))
+                .executePnfRetrieval(anyString(), anyString());
+        verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    private PnfAaiObject constructPnfObject(String pnfName, String hsiCfsServiceInstanceId) {
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/" + hsiCfsServiceInstanceId)
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("customer.global-customer-id")
+                                        .relationshipValue("Demonstration").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-subscription.service-type")
+                                        .relationshipValue("BBS-CFS").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-instance.service-instance-id")
+                                        .relationshipValue(hsiCfsServiceInstanceId).build())
+                        )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        return ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+    }
+
+    private ServiceInstanceAaiObject constructHsiCfsServiceInstanceObject(String hsiCfsServiceInstanceId,
+                                                                          String pnfName,
+                                                                          String rgwMacAddress) {
+        String orchestrationStatus = "active";
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("pnf")
+                        .relatedLink("/pnfs/pnf/" + pnfName)
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("pnf.pnf-name")
+                                .relationshipValue(pnfName).build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
+                ImmutableMetadataEntryAaiObject.builder()
+                        .metaname("rgw-mac-address")
+                        .metavalue(rgwMacAddress)
+                        .build();
+
+        MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
+                .metadataEntries(Collections.singletonList(metadataEntry))
+                .build();
+
+        // Finally construct Service Instance object data
+        return ImmutableServiceInstanceAaiObject.builder()
+                .serviceInstanceId(hsiCfsServiceInstanceId)
+                .orchestrationStatus(orchestrationStatus)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .metadataListAaiObject(metadataListAaiObject)
+                .build();
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..dbd1aab1
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/ReRegistrationPipelineTest.java
@@ -0,0 +1,745 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.pipelines;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_REREGISTRATION_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
+import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.UUID;
+
+import javax.net.ssl.SSLException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.exceptions.AaiTaskException;
+import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
+import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
+import org.onap.bbs.event.processor.model.ImmutableMetadataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableMetadataListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePnfAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePropertyAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableReRegistrationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipDataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableServiceInstanceAaiObject;
+import org.onap.bbs.event.processor.model.MetadataListAaiObject;
+import org.onap.bbs.event.processor.model.PnfAaiObject;
+import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
+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.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+// We can safely suppress unchecked assignment warnings for the ResponseEntity mock
+@SuppressWarnings("unchecked")
+@DisplayName("PNF Re-registration Pipeline Unit-Tests")
+class ReRegistrationPipelineTest {
+
+    private ReRegistrationPipeline pipeline;
+    private ApplicationConfiguration configuration;
+    private DmaapReRegistrationConsumerTask consumerTask;
+    private DmaapPublisherTask publisherTask;
+    private AaiClientTask aaiClientTask;
+
+    private ResponseEntity<String> responseEntity;
+
+    @BeforeEach
+    void setup() {
+
+        responseEntity = Mockito.mock(ResponseEntity.class);
+
+        configuration = Mockito.mock(ApplicationConfiguration.class);
+        consumerTask = Mockito.mock(DmaapReRegistrationConsumerTask.class);
+        publisherTask = Mockito.mock(DmaapPublisherTask.class);
+        aaiClientTask = Mockito.mock(AaiClientTask.class);
+
+        when(configuration.getReRegistrationCloseLoopControlName())
+                .thenReturn("controlName");
+        when(configuration.getReRegistrationCloseLoopPolicyScope())
+                .thenReturn("policyScope");
+
+        pipeline = new ReRegistrationPipeline(configuration, consumerTask,
+                publisherTask, aaiClientTask, new HashMap<>());
+    }
+
+    @Test
+    void handleEmptyResponseFromDmaap() throws SSLException {
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(anyString()))
+                .thenReturn(Flux.error(new EmptyDmaapResponseException("Mock empty")));
+
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @Test
+    void noResponseFromDmaap_PipelineTimesOut() throws SSLException {
+
+        // Prepare mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
+                .thenReturn(Flux.never());
+
+        // Execute pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @Test
+    void noResponseFromAai_PipelineTimesOut() throws SSLException {
+
+        String pnfName = "olt1";
+        String attachmentPoint = "olt2-2-2";
+        String remoteId = "newRemoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        // Prepare mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString())).thenReturn(Mono.never());
+
+        // Execute pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "olt1-1-1", hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
+
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject));
+
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.never());
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObjectWithoutLogicalLink(pnfName, hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
+
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject));
+
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verify(aaiClientTask).executePnfRetrieval(anyString(), anyString());
+        verifyNoMoreInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "old-attachment-point", hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
+
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject));
+
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .attachmentPoint(attachmentPoint1)
+                .remoteId(remoteId1)
+                .cVlan(cvlan1)
+                .sVlan(svlan)
+                .build();
+        ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .attachmentPoint(attachmentPoint2)
+                .remoteId(remoteId2)
+                .cVlan(cvlan2)
+                .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);
+
+        // 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_REREGISTRATION_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(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(publisherTask, times(2)).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @Test
+    void singleEvent_withPnfErrorReply_handleGracefully() throws SSLException {
+
+        String pnfName = "olt1";
+        String attachmentPoint = "olt2-2-2";
+        String remoteId = "newRemoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        // Prepare Mocks
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .verifyComplete();
+
+        verify(aaiClientTask).executePnfRetrieval(anyString(), anyString());
+        verifyNoMoreInteractions(aaiClientTask);
+        verifyZeroInteractions(publisherTask);
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .attachmentPoint(attachmentPoint1)
+                .remoteId(remoteId1)
+                .cVlan(cvlan1)
+                .sVlan(svlan)
+                .build();
+        ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .attachmentPoint(attachmentPoint2)
+                .remoteId(remoteId2)
+                .cVlan(cvlan2)
+                .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);
+
+        // 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_REREGISTRATION_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(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .attachmentPoint(attachmentPoint1)
+                .remoteId(remoteId1)
+                .cVlan(cvlan1)
+                .sVlan(svlan)
+                .build();
+        ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .attachmentPoint(attachmentPoint2)
+                .remoteId(remoteId2)
+                .cVlan(cvlan2)
+                .sVlan(svlan)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName1, "old-attachment-point", hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, cvlan1);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
+                .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.just(pnfAaiObject))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(aaiClientTask, times(2)).executePnfRetrieval(anyString(), anyString());
+        verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    @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();
+
+        // Prepare stubbed replies
+        ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName1)
+                .attachmentPoint(attachmentPoint1)
+                .remoteId(remoteId1)
+                .cVlan(cvlan1)
+                .sVlan(svlan)
+                .build();
+        ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(pnfName2)
+                .attachmentPoint(attachmentPoint2)
+                .remoteId(remoteId2)
+                .cVlan(cvlan2)
+                .sVlan(svlan)
+                .build();
+
+        PnfAaiObject pnfAaiObject = constructPnfObject(pnfName2, "old-attachment-point", hsiCfsServiceInstanceId);
+        ServiceInstanceAaiObject hsiCfsServiceInstance =
+                constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, cvlan2);
+
+        // Prepare Mocks
+        String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
+                hsiCfsServiceInstance.getServiceInstanceId());
+
+        when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
+        when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
+                .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
+        when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
+                .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")))
+                .thenReturn(Mono.just(pnfAaiObject));
+        when(aaiClientTask
+                .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
+                .thenReturn(Mono.just(hsiCfsServiceInstance));
+
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(HttpStatus.OK.value()));
+        when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(responseEntity));
+
+        // Execute the pipeline
+        StepVerifier.create(pipeline.executePipeline())
+                .expectSubscription()
+                .assertNext(r -> assertEquals(HttpStatus.OK, r.getStatusCode()))
+                .verifyComplete();
+
+        verify(aaiClientTask, times(2))
+                .executePnfRetrieval(anyString(), anyString());
+        verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
+        verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
+    }
+
+    private PnfAaiObject constructPnfObject(String pnfName, String attachmentPoint,
+                                            String hsiCfsServiceInstanceId) {
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/" + hsiCfsServiceInstanceId)
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("customer.global-customer-id")
+                                        .relationshipValue("Demonstration").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-subscription.service-type")
+                                        .relationshipValue("BBS-CFS").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-instance.service-instance-id")
+                                        .relationshipValue(hsiCfsServiceInstanceId).build())
+                        )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                        .build();
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject secondRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("logical-link")
+                        .relatedLink("/network/logical-links/logical-link/" + attachmentPoint)
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("logical-link.link-name")
+                                .relationshipValue(attachmentPoint).build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Arrays.asList(firstRelationshipEntry, secondRelationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        return ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+    }
+
+    private PnfAaiObject constructPnfObjectWithoutLogicalLink(String pnfName, String hsiCfsServiceInstanceId) {
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/" + hsiCfsServiceInstanceId)
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("customer.global-customer-id")
+                                        .relationshipValue("Demonstration").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-subscription.service-type")
+                                        .relationshipValue("BBS-CFS").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-instance.service-instance-id")
+                                        .relationshipValue(hsiCfsServiceInstanceId).build())
+                        )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        return ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+    }
+
+    private ServiceInstanceAaiObject constructHsiCfsServiceInstanceObject(String hsiCfsServiceInstanceId,
+                                                                             String pnfName,
+                                                                             String cvlan) {
+        String orchestrationStatus = "active";
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("pnf")
+                        .relatedLink("/pnfs/pnf/" + pnfName)
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("pnf.pnf-name")
+                                .relationshipValue(pnfName).build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
+                ImmutableMetadataEntryAaiObject.builder()
+                        .metaname("cvlan")
+                        .metavalue(cvlan)
+                        .build();
+
+        MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
+                .metadataEntries(Collections.singletonList(metadataEntry))
+                .build();
+
+        // Finally construct Service Instance object data
+        return ImmutableServiceInstanceAaiObject.builder()
+                .serviceInstanceId(hsiCfsServiceInstanceId)
+                .orchestrationStatus(orchestrationStatus)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .metadataListAaiObject(metadataListAaiObject)
+                .build();
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..f721ca7e
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/pipelines/SchedulerTest.java
@@ -0,0 +1,189 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.pipelines;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.ScheduledFuture;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.config.ConsulConfigurationGateway;
+import org.springframework.scheduling.TaskScheduler;
+
+// We can safely suppress unchecked assignment warnings for the ScheduledFuture mock
+@SuppressWarnings("unchecked")
+@DisplayName("Application Task Scheduler Unit-Tests")
+class SchedulerTest {
+
+    private Scheduler applicationScheduler;
+    private ApplicationConfiguration configuration;
+    private TaskScheduler taskScheduler;
+
+    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);
+        this.applicationScheduler = new Scheduler(configuration, configurationGateway, taskScheduler,
+                reRegistrationPipeline, cpeAuthenticationPipeline);
+    }
+
+    @Test
+    void scheduleTasksWithValidSchedulingPeriod_Succeeds() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        ScheduledFuture scheduledFuture = Mockito.mock(ScheduledFuture.class);
+        when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any(Instant.class), any(Duration.class)))
+               .thenReturn(scheduledFuture);
+
+        applicationScheduler.setupScheduler();
+        assertAll("Scheduler with valid Scheduling period",
+            () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(2, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+
+    @Test
+    void cancellingRunningTasksSucceeds_tasksAreDeleted() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        when(scheduledFuture1.cancel(false)).thenReturn(true);
+        when(scheduledFuture2.cancel(false)).thenReturn(true);
+        when(scheduledFuture1.isCancelled()).thenReturn(true);
+        when(scheduledFuture2.isCancelled()).thenReturn(true);
+        when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any(Instant.class), any(Duration.class)))
+                .thenReturn(scheduledFuture1).thenReturn(scheduledFuture2);
+
+        applicationScheduler.setupScheduler();
+        boolean result = applicationScheduler.cancelScheduledProcessingTasks();
+        assertAll("Successfully cancelling tasks",
+            () -> assertTrue(result, "Result of cancellation task"),
+            () -> assertEquals(0, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+
+    @Test
+    void cancellingRunningTasksPartiallyFailing_tasksAreNotDeleted() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        when(scheduledFuture1.cancel(false)).thenReturn(true);
+        when(scheduledFuture2.cancel(false)).thenReturn(false);
+        when(scheduledFuture1.isCancelled()).thenReturn(true);
+        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.cancelScheduledProcessingTasks();
+        assertAll("Partially cancelling tasks",
+            () -> assertFalse(result, "Result of cancellation task"),
+            () -> assertEquals(1, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(1, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+
+    @Test
+    void cancellingRunningTasksFailingForAllOfThem_noTasksAreDeleted() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        when(scheduledFuture1.cancel(false)).thenReturn(false);
+        when(scheduledFuture2.cancel(false)).thenReturn(false);
+        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.cancelScheduledProcessingTasks();
+        assertAll("Failing in cancelling tasks",
+            () -> assertFalse(result, "Result of cancellation task"),
+            () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(2, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+
+    @Test
+    void reSchedulingWithExistingActiveTasks_Fails() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture 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();
+        assertAll("Rescheduling with active tasks",
+            () -> assertFalse(result, "Result of re-scheduling"),
+            () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(2, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+
+    @Test
+    void reSchedulingWithExistingCancelledTasks_Succeeds() {
+        when(configuration.getPipelinesPollingIntervalInSeconds()).thenReturn(20);
+        // Initial tasks
+        ScheduledFuture scheduledFuture1 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture scheduledFuture2 = Mockito.mock(ScheduledFuture.class);
+        // Re-scheduled tasks
+        ScheduledFuture scheduledFuture3 = Mockito.mock(ScheduledFuture.class);
+        ScheduledFuture scheduledFuture4 = Mockito.mock(ScheduledFuture.class);
+        when(scheduledFuture1.isCancelled()).thenReturn(true);
+        when(scheduledFuture2.isCancelled()).thenReturn(true);
+        when(scheduledFuture3.isCancelled()).thenReturn(false);
+        when(scheduledFuture4.isCancelled()).thenReturn(false);
+        when(taskScheduler.scheduleAtFixedRate(any(Runnable.class), any(Instant.class), any(Duration.class)))
+                .thenReturn(scheduledFuture1)
+                .thenReturn(scheduledFuture2)
+                .thenReturn(scheduledFuture3)
+                .thenReturn(scheduledFuture4);
+
+        applicationScheduler.setupScheduler();
+        boolean result = applicationScheduler.reScheduleProcessingTasks();
+        assertAll("Rescheduling with cancelled tasks",
+            () -> assertTrue(result, "Result of re-scheduling"),
+            () -> assertEquals(2, applicationScheduler.numberOfTotalTasks(), "Total tasks"),
+            () -> assertEquals(2, applicationScheduler.numberOfActiveTasks(), "Active tasks"),
+            () -> assertEquals(0, applicationScheduler.numberOfCancelledTasks(), "Cancelled tasks")
+        );
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..db5f7cb1
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/AaiClientTaskImplTest.java
@@ -0,0 +1,239 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.tasks;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapterFactory;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.ServiceLoader;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.exceptions.AaiTaskException;
+import org.onap.bbs.event.processor.model.ImmutableMetadataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableMetadataListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePnfAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePropertyAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipDataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableServiceInstanceAaiObject;
+import org.onap.bbs.event.processor.model.MetadataListAaiObject;
+import org.onap.bbs.event.processor.model.PnfAaiObject;
+import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
+import org.onap.bbs.event.processor.utilities.AaiReactiveClient;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class AaiClientTaskImplTest {
+
+    private AaiReactiveClient reactiveClient;
+
+    private AaiClientTask task;
+
+    @BeforeEach
+    void init() {
+        GsonBuilder gsonBuilder = new GsonBuilder();
+        ServiceLoader.load(TypeAdapterFactory.class).forEach(gsonBuilder::registerTypeAdapterFactory);
+        reactiveClient = Mockito.mock(AaiReactiveClient.class);
+        task = new AaiClientTaskImpl(reactiveClient);
+    }
+
+    @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");
+
+        verify(reactiveClient).getPnfObjectDataFor("some-url");
+        assertNull(pnf.block(Duration.ofSeconds(5)));
+    }
+
+    @Test
+    void passingEmptyServiceInstanceObject_NothingHappens() throws AaiTaskException {
+        when(reactiveClient.getServiceInstanceObjectDataFor(any(String.class))).thenReturn(Mono.empty());
+        Mono<ServiceInstanceAaiObject> serviceInstance =
+                task.executeServiceInstanceRetrieval("Empty Service Instance task", "some-url");
+
+        verify(reactiveClient).getServiceInstanceObjectDataFor("some-url");
+        assertNull(serviceInstance.block(Duration.ofSeconds(5)));
+    }
+
+    @Test
+    void passingPnfObject_taskSucceeds() throws AaiTaskException {
+
+        String pnfName = "pnf-1";
+        String attachmentPoint = "olt1-1-1";
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS/service-instances"
+                                + "/service-instance/84003b26-6b76-4c75-b805-7b14ab4ffaef")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("customer.global-customer-id")
+                                        .relationshipValue("Demonstration").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-subscription.service-type")
+                                        .relationshipValue("BBS").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-instance.service-instance-id")
+                                        .relationshipValue("84003b26-6b76-4c75-b805-7b14ab4ffaef").build())
+                        )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                        .build();
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject secondRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("logical-link")
+                        .relatedLink("/network/logical-links/logical-link/" + attachmentPoint)
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("logical-link.link-name")
+                                .relationshipValue(attachmentPoint).build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Arrays.asList(firstRelationshipEntry, secondRelationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        PnfAaiObject pnfAaiObject = ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+
+        when(reactiveClient.getPnfObjectDataFor(any(String.class))).thenReturn(Mono.just(pnfAaiObject));
+        Mono<PnfAaiObject> pnf = task.executePnfRetrieval("Normal PNF retrieval task", "some-url");
+
+        verify(reactiveClient).getPnfObjectDataFor("some-url");
+        assertNotNull(pnf.block(Duration.ofSeconds(5)));
+
+        StepVerifier.create(pnf)
+                .expectSubscription()
+                .consumeNextWith(aPnf -> {
+                    Assertions.assertEquals(pnfName, aPnf.getPnfName(), "PNF Name in response does not match");
+                    String extractedAttachmentPoint = aPnf.getRelationshipListAaiObject().getRelationshipEntries()
+                            .stream()
+                            .filter(e -> e.getRelatedTo().equals("logical-link"))
+                            .flatMap(e -> e.getRelationshipData().stream())
+                            .filter(d -> d.getRelationshipKey().equals("logical-link.link-name"))
+                            .map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
+                            .findFirst().orElseThrow(AaiClientTaskTestException::new);
+                    Assertions.assertEquals(attachmentPoint, extractedAttachmentPoint,
+                            "Attachment point in response does not match");
+                })
+                .verifyComplete();
+    }
+
+    @Test
+    void passingServiceInstanceObject_taskSucceeds() throws AaiTaskException {
+
+        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        String orchestrationStatus = "active";
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/bb374844-44e4-488f-8381-fb5a0e3e6989")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("service-instance.service-instance-id")
+                                .relationshipValue("bb374844-44e4-488f-8381-fb5a0e3e6989").build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
+                ImmutableMetadataEntryAaiObject.builder()
+                        .metaname("cvlan")
+                        .metavalue("1005")
+                        .build();
+
+        MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
+                .metadataEntries(Collections.singletonList(metadataEntry))
+                .build();
+
+        // Finally construct Service Instance object data
+        ServiceInstanceAaiObject serviceInstanceAaiObject = ImmutableServiceInstanceAaiObject.builder()
+                .serviceInstanceId(serviceInstanceId)
+                .orchestrationStatus(orchestrationStatus)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .metadataListAaiObject(metadataListAaiObject)
+                .build();
+
+        when(reactiveClient.getServiceInstanceObjectDataFor(any(String.class)))
+                .thenReturn(Mono.just(serviceInstanceAaiObject));
+        Mono<ServiceInstanceAaiObject> serviceInstance =
+                task.executeServiceInstanceRetrieval("Normal Service Instance retrieval task",
+                        "some-url");
+
+        verify(reactiveClient).getServiceInstanceObjectDataFor("some-url");
+        assertNotNull(serviceInstance.block(Duration.ofSeconds(5)));
+
+        StepVerifier.create(serviceInstance)
+                .expectSubscription()
+                .consumeNextWith(instance -> {
+                    Assertions.assertEquals(serviceInstanceId, instance.getServiceInstanceId(),
+                            "Service Instance ID in response does not match");
+
+                    MetadataListAaiObject extractedMetadataListObject =
+                            instance.getMetadataListAaiObject().orElseThrow(AaiClientTaskTestException::new);
+
+                    MetadataListAaiObject.MetadataEntryAaiObject extractedMetadataEntry =
+                            extractedMetadataListObject.getMetadataEntries()
+                                    .stream()
+                                    .filter(m -> m.getMetaname().equals("cvlan"))
+                                    .findFirst().orElseThrow(AaiClientTaskTestException::new);
+
+                    Assertions.assertEquals("1005", extractedMetadataEntry.getMetavalue(),
+                            "CVLAN in response does not match");
+                })
+                .verifyComplete();
+    }
+
+    private static class AaiClientTaskTestException extends RuntimeException {}
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..538ff1de
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapCpeAuthenticationConsumerTaskImplTest.java
@@ -0,0 +1,149 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.tasks;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+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.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 reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class DmaapCpeAuthenticationConsumerTaskImplTest {
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
+            + "\"stateChangeFields\": {"
+            + " \"oldState\": \"%s\","
+            + " \"newState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    private static DmaapCpeAuthenticationConsumerTask dmaapConsumerTask;
+    private static CpeAuthenticationConsumerDmaapModel cpeAuthenticationConsumerDmaapModel;
+    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
+    private static String eventsArray;
+
+    @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";
+
+        // 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);
+
+        cpeAuthenticationConsumerDmaapModel = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(sourceName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        String 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 {
+        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(""));
+
+        StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
+                .expectSubscription()
+                .expectError(EmptyDmaapResponseException.class);
+        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse();
+    }
+
+    @Test
+    void passingNormalMessage_ResponseSucceeds() throws Exception {
+        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(eventsArray));
+
+        StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
+                .expectSubscription()
+                .consumeNextWith(e -> Assert.assertEquals(e, cpeAuthenticationConsumerDmaapModel));
+        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse();
+    }
+
+    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();
+    }
+}
\ 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
new file mode 100644
index 00000000..199a43ec
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapPublisherTaskImplTest.java
@@ -0,0 +1,173 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.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;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+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.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.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 org.springframework.http.ResponseEntity;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class DmaapPublisherTaskImplTest {
+
+    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 =
+                "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 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");
+
+        controlLoopPublisherDmaapModel = ImmutableControlLoopPublisherDmaapModel.builder()
+                .closedLoopEventClient(closedLoopEventClient)
+                .policyVersion(policyVersion)
+                .policyName(policyName)
+                .policyScope(policyScope)
+                .targetType(targetType)
+                .aaiEnrichmentData(aaiEnrichmentData)
+                .closedLoopAlarmStart(closedLoopAlarmStart)
+                .closedLoopEventStatus(closedLoopEventStatus)
+                .closedLoopControlName(closedLoopControlName)
+                .version(version)
+                .target(target)
+                .requestId(requestId)
+                .originator(from)
+                .build();
+
+        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 {
+        ResponseEntity<String> responseEntity = setupMocks(HttpStatus.OK.value());
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.OK);
+        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel)).expectSubscription()
+                .expectNext(responseEntity).verifyComplete();
+
+        verify(reactiveHttpClient, times(1))
+                .getDMaaPProducerResponse(controlLoopPublisherDmaapModel);
+        verifyNoMoreInteractions(reactiveHttpClient);
+    }
+
+    @Test
+    void passingNormalMessage_IncorrectResponseIsHandled() throws DmaapException {
+        ResponseEntity<String> responseEntity = setupMocks(HttpStatus.UNAUTHORIZED.value());
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.UNAUTHORIZED);
+        StepVerifier.create(task.execute(controlLoopPublisherDmaapModel)).expectSubscription()
+                .expectNext(responseEntity).verifyComplete();
+
+        verify(reactiveHttpClient, times(1))
+                .getDMaaPProducerResponse(controlLoopPublisherDmaapModel);
+        verifyNoMoreInteractions(reactiveHttpClient);
+    }
+
+    // We can safely suppress unchecked assignment warning here since it is a mock class
+    @SuppressWarnings("unchecked")
+    private ResponseEntity<String> setupMocks(Integer httpResponseCode) {
+
+        ResponseEntity<String> responseEntity = mock(ResponseEntity.class);
+        when(responseEntity.getStatusCode()).thenReturn(HttpStatus.valueOf(httpResponseCode));
+
+        reactiveHttpClient = mock(DMaaPPublisherReactiveHttpClient.class);
+        when(reactiveHttpClient.getDMaaPProducerResponse(any()))
+                .thenReturn(Mono.just(responseEntity));
+
+        PublisherReactiveHttpClientFactory httpClientFactory = mock(PublisherReactiveHttpClientFactory.class);
+        doReturn(reactiveHttpClient).when(httpClientFactory).create(dmaapPublisherConfiguration);
+
+        task = new DmaapPublisherTaskImpl(configuration, httpClientFactory);
+
+        return responseEntity;
+    }
+
+    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();
+    }
+}
\ 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
new file mode 100644
index 00000000..c9a461d8
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/tasks/DmaapReRegistrationConsumerTaskImplTest.java
@@ -0,0 +1,147 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.tasks;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+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.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 reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class DmaapReRegistrationConsumerTaskImplTest {
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
+            + "\"additionalFields\": {"
+            + " \"attachment-point\": \"%s\","
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}}";
+
+    private static DmaapReRegistrationConsumerTaskImpl dmaapConsumerTask;
+    private static ReRegistrationConsumerDmaapModel reRegistrationConsumerDmaapModel;
+    private static DMaaPConsumerReactiveHttpClient dMaaPConsumerReactiveHttpClient;
+    private static String message;
+
+    @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";
+
+        // 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);
+
+        reRegistrationConsumerDmaapModel = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(sourceName)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        message = String.format("[" + RE_REGISTRATION_EVENT_TEMPLATE + "]",
+                sourceName,
+                attachmentPoint,
+                remoteId,
+                cvlan,
+                svlan);
+    }
+
+    @AfterEach
+    void resetMock() {
+        reset(dMaaPConsumerReactiveHttpClient);
+    }
+
+    @Test
+    void passingEmptyMessage_NothingHappens() throws Exception {
+        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(""));
+
+        StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
+                .expectSubscription()
+                .expectError(EmptyDmaapResponseException.class);
+        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse();
+    }
+
+    @Test
+    void passingNormalMessage_ResponseSucceeds() throws Exception {
+        when(dMaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse()).thenReturn(Mono.just(message));
+
+        StepVerifier.create(dmaapConsumerTask.execute("Sample input"))
+                .expectSubscription()
+                .consumeNextWith(e -> Assert.assertEquals(e, reRegistrationConsumerDmaapModel));
+        verify(dMaaPConsumerReactiveHttpClient).getDMaaPConsumerResponse();
+    }
+
+    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();
+    }
+}
\ 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
new file mode 100644
index 00000000..8e3c46ba
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/AaiReactiveClientTest.java
@@ -0,0 +1,257 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.utilities;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.givenThat;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static org.mockito.Mockito.when;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapterFactory;
+
+import java.util.Arrays;
+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;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.config.AaiClientConfiguration;
+import org.onap.bbs.event.processor.config.ApplicationConfiguration;
+import org.onap.bbs.event.processor.model.ImmutableMetadataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableMetadataListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePnfAaiObject;
+import org.onap.bbs.event.processor.model.ImmutablePropertyAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipDataEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipEntryAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableRelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ImmutableServiceInstanceAaiObject;
+import org.onap.bbs.event.processor.model.MetadataListAaiObject;
+import org.onap.bbs.event.processor.model.PnfAaiObject;
+import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
+import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import reactor.test.StepVerifier;
+
+@ExtendWith(SpringExtension.class)
+class AaiReactiveClientTest {
+
+    private static final int PORT = 9999;
+
+    private static AaiReactiveClient reactiveClient;
+    private static Gson gson;
+    private static WireMockServer wireMockServer;
+
+    @BeforeAll
+    static void init() throws SSLException {
+        GsonBuilder 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);
+        when(configuration.getAaiClientConfiguration()).thenReturn(aaiClientConfiguration);
+        when(aaiClientConfiguration.aaiUserName()).thenReturn("AAI");
+        when(aaiClientConfiguration.aaiUserPassword()).thenReturn("AAI");
+        when(aaiClientConfiguration.aaiHeaders()).thenReturn(new HashMap<>());
+        when(aaiClientConfiguration.enableAaiCertAuth()).thenReturn(false);
+
+        reactiveClient = new AaiReactiveClient(configuration, gson);
+
+        wireMockServer = new WireMockServer(PORT);
+        WireMock.configureFor("localhost", PORT);
+    }
+
+    @BeforeEach
+    void wireMockSetup() {
+        wireMockServer.start();
+    }
+
+    @AfterEach
+    void wireMockTearDown() {
+        wireMockServer.start();
+    }
+
+    @Test
+    void sendingReactiveRequestForPnf_Succeeds() {
+
+        String pnfName = "pnf-1";
+        String attachmentPoint = "olt1-1-1";
+
+        String pnfUrl = String.format("/aai/v14/network/pnfs/pnf/%s?depth=1", pnfName);
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS/service-instances"
+                                + "/service-instance/84003b26-6b76-4c75-b805-7b14ab4ffaef")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Arrays.asList(
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("customer.global-customer-id")
+                                        .relationshipValue("Demonstration").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-subscription.service-type")
+                                        .relationshipValue("BBS").build(),
+                                ImmutableRelationshipDataEntryAaiObject.builder()
+                                        .relationshipKey("service-instance.service-instance-id")
+                                        .relationshipValue("84003b26-6b76-4c75-b805-7b14ab4ffaef").build())
+                        )
+                        .relatedToProperties(Collections.singletonList(
+                                ImmutablePropertyAaiObject.builder()
+                                        .propertyKey("service-instance.service-instance-name")
+                                        .propertyValue("bbs-instance").build())
+                        )
+                        .build();
+
+        RelationshipListAaiObject.RelationshipEntryAaiObject secondRelationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("logical-link")
+                        .relatedLink("/network/logical-links/logical-link/" + attachmentPoint)
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("logical-link.link-name")
+                                .relationshipValue(attachmentPoint).build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Arrays.asList(firstRelationshipEntry, secondRelationshipEntry))
+                .build();
+
+        // Finally construct PNF object data
+        PnfAaiObject pnfAaiObject = ImmutablePnfAaiObject.builder()
+                .pnfName(pnfName)
+                .isInMaintenance(true)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .build();
+
+        givenThat(get(urlEqualTo(pnfUrl))
+                .willReturn(aResponse().withStatus(200)
+                        .withHeader("Content-Type", "application/json")
+                        .withBody(gson.toJson(pnfAaiObject, ImmutablePnfAaiObject.class))));
+
+        StepVerifier.create(reactiveClient.getPnfObjectDataFor("http://127.0.0.1:" + PORT + pnfUrl))
+                .expectSubscription()
+                .consumeNextWith(pnf -> {
+                    Assertions.assertEquals(pnfName, pnf.getPnfName(), "PNF Name in response does not match");
+                    String extractedAttachmentPoint = pnf.getRelationshipListAaiObject().getRelationshipEntries()
+                            .stream()
+                            .filter(e -> e.getRelatedTo().equals("logical-link"))
+                            .flatMap(e -> e.getRelationshipData().stream())
+                            .filter(d -> d.getRelationshipKey().equals("logical-link.link-name"))
+                            .map(RelationshipListAaiObject.RelationshipDataEntryAaiObject::getRelationshipValue)
+                            .findFirst().orElseThrow(AaiReactiveClientTestException::new);
+                    Assertions.assertEquals(attachmentPoint, extractedAttachmentPoint,
+                            "Attachment point in response does not match");
+                })
+                .verifyComplete();
+    }
+
+    @Test
+    void sendingReactiveRequestForServiceInstance_Succeeds() {
+
+        String serviceInstanceId = "84003b26-6b76-4c75-b805-7b14ab4ffaef";
+        String orchestrationStatus = "active";
+
+        String serviceInstanceUrl =
+                String.format("/aai/v14/nodes/service-instances/service-instance/%s?format=resource_and_url",
+                serviceInstanceId);
+
+        // Build Relationship Data
+        RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
+                ImmutableRelationshipEntryAaiObject.builder()
+                        .relatedTo("service-instance")
+                        .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
+                                + "/service-subscription/BBS-CFS/service-instances"
+                                + "/service-instance/bb374844-44e4-488f-8381-fb5a0e3e6989")
+                        .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
+                        .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
+                                .relationshipKey("service-instance.service-instance-id")
+                                .relationshipValue("bb374844-44e4-488f-8381-fb5a0e3e6989").build()))
+                        .build();
+
+        RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
+                .relationshipEntries(Collections.singletonList(relationshipEntry))
+                .build();
+
+        MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
+                ImmutableMetadataEntryAaiObject.builder()
+                        .metaname("cvlan")
+                        .metavalue("1005")
+                        .build();
+
+        MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
+                .metadataEntries(Collections.singletonList(metadataEntry))
+                .build();
+
+        // Finally construct Service Instance object data
+        ServiceInstanceAaiObject serviceInstanceAaiObject = ImmutableServiceInstanceAaiObject.builder()
+                .serviceInstanceId(serviceInstanceId)
+                .orchestrationStatus(orchestrationStatus)
+                .relationshipListAaiObject(relationshipListAaiObject)
+                .metadataListAaiObject(metadataListAaiObject)
+                .build();
+
+        givenThat(get(urlEqualTo(serviceInstanceUrl))
+                .willReturn(aResponse().withStatus(200)
+                        .withHeader("Content-Type", "application/json")
+                        .withBody(gson.toJson(serviceInstanceAaiObject, ImmutableServiceInstanceAaiObject.class))));
+
+        StepVerifier.create(
+                reactiveClient.getServiceInstanceObjectDataFor("http://127.0.0.1:" + PORT + serviceInstanceUrl)
+        )
+                .expectSubscription()
+                .consumeNextWith(serviceInstance -> {
+                    Assertions.assertEquals(serviceInstanceId, serviceInstance.getServiceInstanceId(),
+                            "Service Instance ID in response does not match");
+
+                    MetadataListAaiObject extractedMetadataListObject =
+                            serviceInstance.getMetadataListAaiObject().orElseThrow(AaiReactiveClientTestException::new);
+
+                    MetadataListAaiObject.MetadataEntryAaiObject extractedMetadataEntry =
+                            extractedMetadataListObject.getMetadataEntries()
+                            .stream()
+                            .filter(m -> m.getMetaname().equals("cvlan"))
+                            .findFirst().orElseThrow(AaiReactiveClientTestException::new);
+
+                    Assertions.assertEquals("1005", extractedMetadataEntry.getMetavalue(),
+                            "CVLAN in response does not match");
+                })
+                .verifyComplete();
+    }
+
+    private static class AaiReactiveClientTestException extends RuntimeException {}
+
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..4ca61f5e
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/CpeAuthenticationDmaapConsumerJsonParserTest.java
@@ -0,0 +1,311 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.utilities;
+
+import static org.mockito.Mockito.spy;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+
+import java.util.Optional;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.model.CpeAuthenticationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.ImmutableCpeAuthenticationConsumerDmaapModel;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class CpeAuthenticationDmaapConsumerJsonParserTest {
+
+    private static JsonParser jsonParser;
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
+            + "\"stateChangeFields\": {"
+            + " \"oldState\": \"%s\","
+            + " \"newState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_AUTHENTICATION_STATE = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"%s\"},"
+            + "\"stateChangeFields\": {"
+            + " \"oldState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME = "{\"event\": {"
+            + "\"commonEventHeader\": { },"
+            + "\"stateChangeFields\": {"
+            + " \"oldState\": \"%s\","
+            + " \"newState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_SOURCE_NAME_VALUE = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"\"},"
+            + "\"stateChangeFields\": {"
+            + " \"oldState\": \"%s\","
+            + " \"newState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    private static final String CPE_AUTHENTICATION_EVENT_TEMPLATE_WITH_MISSING_STATE_CHANGE_FIELDS = "{\"event\": {"
+            + "\"commonEventHeader\": { \"sourceName\":\"\"},"
+            + "\"somethingElse\": {"
+            + " \"oldState\": \"%s\","
+            + " \"newState\": \"%s\","
+            + " \"stateInterface\": \"%s\","
+            + " \"additionalFields\": {"
+            + "   \"macAddress\": \"%s\","
+            + "   \"swVersion\": \"%s\""
+            + "}}}}";
+
+    @BeforeAll
+    static void init() {
+        jsonParser = new JsonParser();
+    }
+
+    @Test
+    void passingNonJson_EmptyFluxIsReturned() {
+
+        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
+                new CpeAuthenticationDmaapConsumerJsonParser();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just("not JSON")))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingNoEvents_EmptyFluxIsReturned() {
+
+        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
+                new CpeAuthenticationDmaapConsumerJsonParser();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just("[]")))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @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";
+
+        String event = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName, oldAuthenticationState,
+                newAuthenticationState, stateInterface, rgwMacAddress, swVersion);
+
+        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
+                spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        JsonElement jsonElement = jsonParser.parse(event);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        CpeAuthenticationConsumerDmaapModel expectedEventObject = ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(sourceName)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress)
+                .swVersion(swVersion)
+                .build();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .expectNext(expectedEventObject);
+    }
+
+    @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";
+
+        String firstEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName1, oldAuthenticationState,
+                newAuthenticationState, stateInterface, rgwMacAddress1, swVersion);
+        String secondEvent = String.format(CPE_AUTHENTICATION_EVENT_TEMPLATE, sourceName2, oldAuthenticationState,
+                newAuthenticationState, stateInterface, rgwMacAddress2, swVersion);
+
+        CpeAuthenticationDmaapConsumerJsonParser consumerJsonParser =
+                spy(new CpeAuthenticationDmaapConsumerJsonParser());
+        JsonElement jsonElement1 = jsonParser.parse(firstEvent);
+        Mockito.doReturn(Optional.of(jsonElement1.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement1);
+        JsonElement jsonElement2 = jsonParser.parse(secondEvent);
+        Mockito.doReturn(Optional.of(jsonElement2.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement2);
+
+        String eventsArray = "[" + firstEvent + secondEvent + "]";
+
+        CpeAuthenticationConsumerDmaapModel expectedFirstEventObject =
+                ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(sourceName1)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress1)
+                .swVersion(swVersion)
+                .build();
+        CpeAuthenticationConsumerDmaapModel expectedSecondEventObject =
+                ImmutableCpeAuthenticationConsumerDmaapModel.builder()
+                .correlationId(sourceName2)
+                .oldAuthenticationState(oldAuthenticationState)
+                .newAuthenticationState(newAuthenticationState)
+                .stateInterface(stateInterface)
+                .rgwMacAddress(rgwMacAddress2)
+                .swVersion(swVersion)
+                .build();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .expectNext(expectedFirstEventObject)
+                .expectNext(expectedSecondEventObject);
+    }
+
+    @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";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @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";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @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";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @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";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+}
\ No newline at end of file
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
new file mode 100644
index 00000000..ca448dd0
--- /dev/null
+++ b/components/bbs-event-processor/src/test/java/org/onap/bbs/event/processor/utilities/ReRegistrationDmaapConsumerJsonParserTest.java
@@ -0,0 +1,297 @@
+/*
+ * ============LICENSE_START=======================================================
+ * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
+ * ================================================================================
+ * Copyright (C) 2019 NOKIA Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.bbs.event.processor.utilities;
+
+import static org.mockito.Mockito.spy;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParser;
+
+import java.util.Optional;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.onap.bbs.event.processor.model.ImmutableReRegistrationConsumerDmaapModel;
+import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
+
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+class ReRegistrationDmaapConsumerJsonParserTest {
+
+    private static JsonParser jsonParser;
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE = "{"
+            + "\"correlationId\":\"%s\","
+            + "\"additionalFields\": {"
+            + " \"attachment-point\": \"%s\","
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}";
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ATTACHMENT_POINT = "{"
+            + "\"correlationId\":\"%s\","
+            + "\"additionalFields\": {"
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}";
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID = "{"
+            + "\"additionalFields\": {"
+            + " \"attachment-point\": \"%s\","
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}";
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID_VALUE = "{"
+            + "\"correlationId\":\"\","
+            + "\"additionalFields\": {"
+            + " \"attachment-point\": \"%s\","
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}";
+
+    private static final String RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ADDITIONAL_FIELDS = "{"
+            + "\"correlationId\":\"%s\","
+            + "\"somethingElse\": {"
+            + " \"attachment-point\": \"%s\","
+            + " \"remote-id\": \"%s\","
+            + " \"cvlan\": \"%s\","
+            + " \"svlan\": \"%s\""
+            + "}}";
+
+    @BeforeAll
+    static void init() {
+        jsonParser = new JsonParser();
+    }
+
+    @Test
+    void passingNonJson_EmptyFluxIsReturned() {
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just("not JSON")))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingNoEvents_EmptyFluxIsReturned() {
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = new ReRegistrationDmaapConsumerJsonParser();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just("[]")))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingOneCorrectEvent_validationSucceeds() {
+
+        String correlationId = "PNF-CorrelationId";
+        String attachmentPoint = "olt1/1/1";
+        String remoteId = "remoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId, attachmentPoint,
+                remoteId, cvlan, svlan);
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        JsonElement jsonElement = jsonParser.parse(event);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        ReRegistrationConsumerDmaapModel expectedEventObject = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(correlationId)
+                .attachmentPoint(attachmentPoint)
+                .remoteId(remoteId)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .expectNext(expectedEventObject);
+    }
+
+    @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";
+
+        String firstEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
+                remoteId1, cvlan, svlan);
+        String secondEvent = String.format(RE_REGISTRATION_EVENT_TEMPLATE, correlationId1, attachmentPoint1,
+                remoteId1, cvlan, svlan);
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        JsonElement jsonElement1 = jsonParser.parse(firstEvent);
+        Mockito.doReturn(Optional.of(jsonElement1.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement1);
+        JsonElement jsonElement2 = jsonParser.parse(secondEvent);
+        Mockito.doReturn(Optional.of(jsonElement2.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement2);
+
+        String eventsArray = "[" + firstEvent + secondEvent + "]";
+
+        ReRegistrationConsumerDmaapModel expectedFirstEventObject = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(correlationId1)
+                .attachmentPoint(attachmentPoint1)
+                .remoteId(remoteId1)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+        ReRegistrationConsumerDmaapModel expectedSecondEventObject = ImmutableReRegistrationConsumerDmaapModel.builder()
+                .correlationId(correlationId2)
+                .attachmentPoint(attachmentPoint2)
+                .remoteId(remoteId2)
+                .cVlan(cvlan)
+                .sVlan(svlan)
+                .build();
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .expectNext(expectedFirstEventObject)
+                .expectNext(expectedSecondEventObject);
+    }
+
+    @Test
+    void passingJsonWithMissingAttachmentPoint_validationFails() {
+
+        String correlationId = "PNF-CorrelationId";
+        String remoteId = "remoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_ATTACHMENT_POINT,
+                correlationId,
+                remoteId,
+                cvlan,
+                svlan);
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        JsonElement jsonElement = jsonParser.parse(event);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingJsonWithMissingCorrelationId_validationFails() {
+
+        String attachmentPoint = "olt1/1/1";
+        String remoteId = "remoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String event = String.format(RE_REGISTRATION_EVENT_TEMPLATE_MISSING_CORRELATION_ID,
+                attachmentPoint,
+                remoteId,
+                cvlan,
+                svlan);
+
+        ReRegistrationDmaapConsumerJsonParser consumerJsonParser = spy(new ReRegistrationDmaapConsumerJsonParser());
+        JsonElement jsonElement = jsonParser.parse(event);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingJsonWithMissingCorrelationIdValue_validationFails() {
+
+        String attachmentPoint = "olt1/1/1";
+        String remoteId = "remoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+    @Test
+    void passingJsonWithMissingAdditionalFields_validationFails() {
+
+        String correlationId = "PNF-CorrelationId";
+        String attachmentPoint = "olt1/1/1";
+        String remoteId = "remoteId";
+        String cvlan = "1005";
+        String svlan = "100";
+
+        String 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);
+        Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
+                .when(consumerJsonParser).getJsonObjectFromAnArray(jsonElement);
+
+        String eventsArray = "[" + event + "]";
+
+        StepVerifier.create(consumerJsonParser.extractModelFromDmaap(Mono.just(eventsArray)))
+                .expectSubscription()
+                .verifyComplete();
+    }
+
+}
\ No newline at end of file