Xacml PDP Register/Unregister Changes

1) Modified activator
	- Extend and use ServiceManagerContainer
	- add topicSink and topicSource properties to TopicEndpoint
	manager
2) Added PAP registration class
3) Added PDP message formatting class
4) Added PDP State Change listener to listen for state change and send
back PDP Status.. then start heartbeat if not started already
5) Added Heartbeat publisher code to send heartbeat status on periodic
interval
6) UPDATE: Made changes to satisfy Jim's review comments
7) Added code to XacmlPdpApplicationManager to get supportedPolicyTypes
8) UPDATE: more changes to satisfy Jim's latest review comments
9) Fixed the failed Junits causing build failure
10) Made change to use NetworkUtil for getting PdpStatus name value per
Jorge's comment

Still need to in separate review:
1) Write Junits for new code

Registration and heartbeat flow as I understand it:
1) PDP starts and sends initial status to PAP with PASSIVE state
2) PDP listens for State Change from the PAP
3) PDP sends updated Active state to the AP
4) PDP begins sending periodic heartbeat status to the PAP

Change-Id: Iafa06ae3f1ee195c7da3710a12d9c095c5413ec2
Issue-ID: POLICY-1449
Signed-off-by: Michael Mokry <michael.mokry@att.com>

17 files changed, 600 insertions, 96 deletions

diff --git a/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpHeartbeatPublisher.java b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpHeartbeatPublisher.java
new file mode 100644
index 00000000..fe730208
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpHeartbeatPublisher.java
@@ -0,0 +1,83 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pdpx.main.comm;
+
+import java.util.Timer;
+import java.util.TimerTask;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class XacmlPdpHeartbeatPublisher extends TimerTask {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpHeartbeatPublisher.class);
+
+    private TopicSinkClient topicSinkClient;
+    private Timer timer;
+    private XacmlPdpMessage heartbeatMessage;
+    private PdpState pdpState;
+
+    private static volatile  boolean alive = false;
+
+    /**
+     * Constructor for instantiating XacmlPdpHeartbeatPublisher.
+     *
+     * @param state of the PDP
+     * @param topicSinkClient used to send heartbeat message
+     */
+    public XacmlPdpHeartbeatPublisher(TopicSinkClient topicSinkClient, PdpState state) {
+        this.topicSinkClient = topicSinkClient;
+        this.heartbeatMessage = new XacmlPdpMessage();
+        this.pdpState = state;
+        timer = new Timer(false);
+        timer.scheduleAtFixedRate(this, 0, 60000); // time interval temp hard coded now but will be parameterized
+        setAlive(true);
+    }
+
+    @Override
+    public void run() {
+        topicSinkClient.send(heartbeatMessage.formatStatusMessage(pdpState));
+        LOGGER.info("Sending Xacml PDP heartbeat to the PAP");
+    }
+
+    /**
+     * Method to terminate the heartbeat.
+     */
+    public void terminate() {
+        timer.cancel();
+        timer.purge();
+        setAlive(false);
+    }
+
+    public void updateInternalState(PdpState state) {
+        this.pdpState = state;
+    }
+
+    public static boolean isAlive() {
+        return alive;
+    }
+
+    public void setAlive(boolean alive) {
+        this.alive = alive;
+    }
+
+}
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpMessage.java b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpMessage.java
new file mode 100644
index 00000000..e05120d7
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpMessage.java
@@ -0,0 +1,64 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pdpx.main.comm;
+
+import java.net.UnknownHostException;
+import org.onap.policy.common.utils.network.NetworkUtil;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.enums.PdpHealthStatus;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.pdpx.main.rest.XacmlPdpApplicationManager;
+import org.onap.policy.pdpx.main.startstop.XacmlPdpActivator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class XacmlPdpMessage {
+
+    // The logger for this class
+    private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpMessage.class);
+
+    /**
+     * Method used to format the status message.
+     *
+     * @param state of the PDP
+     * @return status message of the PDP
+     * @throws UnknownHostException if cannot get hostname
+     */
+    public PdpStatus formatStatusMessage(PdpState state) {
+        PdpStatus status = new PdpStatus();
+        status.setName(NetworkUtil.getHostname());
+
+        if (XacmlPdpActivator.getCurrent().isAlive()) {
+            status.setHealthy(PdpHealthStatus.HEALTHY);
+        } else {
+            status.setHealthy(PdpHealthStatus.NOT_HEALTHY);
+        }
+
+        status.setPdpType("xacml");
+        status.setState(state);
+        status.setSupportedPolicyTypes(XacmlPdpApplicationManager.getToscaPolicyTypeIdents());
+
+        return status;
+
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpPapRegistration.java b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpPapRegistration.java
new file mode 100644
index 00000000..b7421964
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pdpx/main/comm/XacmlPdpPapRegistration.java
@@ -0,0 +1,60 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pdpx.main.comm;
+
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class XacmlPdpPapRegistration {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpPapRegistration.class);
+    private final TopicSinkClient client;
+
+    /**
+     * Constructs the object.
+     * @param client name of the TopickSinkClient
+     */
+    public XacmlPdpPapRegistration(TopicSinkClient client) {
+        this.client = client;
+    }
+
+    /**
+     * Sends PDP register and unregister message to the PAP.
+     * @param status of the PDP
+     * @throws TopicSinkClientException if the topic sink does not exist
+     */
+    public void pdpRegistration(PdpStatus status) throws TopicSinkClientException {
+        try {
+            if (!client.send(status)) {
+                LOGGER.error("Failed to send to topic sink " + client.getTopic());
+                return;
+            }
+        } catch (IllegalStateException e) {
+            LOGGER.error("Failed ot send to topic sink " + client.getTopic(), e);
+            return;
+        }
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/comm/listeners/XacmlPdpStateChangeListener.java b/main/src/main/java/org/onap/policy/pdpx/main/comm/listeners/XacmlPdpStateChangeListener.java
new file mode 100644
index 00000000..68bf06dd
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pdpx/main/comm/listeners/XacmlPdpStateChangeListener.java
@@ -0,0 +1,81 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 AT&T 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pdpx.main.comm.listeners;
+
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
+import org.onap.policy.common.endpoints.listeners.ScoListener;
+import org.onap.policy.common.utils.coder.StandardCoderObject;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.pdpx.main.comm.XacmlPdpHeartbeatPublisher;
+import org.onap.policy.pdpx.main.comm.XacmlPdpMessage;
+import org.onap.policy.pdpx.main.startstop.XacmlPdpActivator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class XacmlPdpStateChangeListener extends ScoListener<PdpStateChange> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpStateChangeListener.class);
+
+    private TopicSinkClient client;
+
+    private XacmlPdpHeartbeatPublisher heartbeat;
+
+    /**
+     * Constructs the object.
+     *
+     * @param client used to send back response after receiving state change message
+     */
+    public XacmlPdpStateChangeListener(TopicSinkClient client) {
+        super(PdpStateChange.class);
+        heartbeat = new XacmlPdpHeartbeatPublisher(client, PdpState.PASSIVE);
+        this.client = client;
+    }
+
+    @Override
+    public void onTopicEvent(CommInfrastructure infra, String topic, StandardCoderObject sco, PdpStateChange message) {
+
+        XacmlPdpMessage newMessage = new XacmlPdpMessage();
+        try {
+            PdpStatus newStatus = newMessage.formatStatusMessage(message.getState());
+
+            // Send State Change Status to PAP
+            if (!client.send(newStatus)) {
+                LOGGER.error("failed to send to topic sink " + client.getTopic());
+                throw new TopicSinkClientException("failed to send to topic sink " + client.getTopic());
+            }
+
+            // Update the heartbeat internal state if publisher is running else create new publisher
+            if (XacmlPdpHeartbeatPublisher.isAlive()) {
+                heartbeat.updateInternalState(message.getState());
+            } else {
+                heartbeat = new XacmlPdpHeartbeatPublisher(client, message.getState());
+            }
+
+        } catch (final Exception e) {
+            LOGGER.error("failed to handle the PDP State Change message.", e);
+        }
+
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpApplicationManager.java b/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpApplicationManager.java
index c97166c1..7d3292a6 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpApplicationManager.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpApplicationManager.java
@@ -22,12 +22,13 @@ package org.onap.policy.pdpx.main.rest;
 
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
-
 import org.onap.policy.models.decisions.concepts.DecisionRequest;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyTypeIdentifier;
 import org.onap.policy.pdp.xacml.application.common.XacmlApplicationServiceProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,6 +39,7 @@ public class XacmlPdpApplicationManager {
     private static boolean needsInit = true;
     private static ServiceLoader<XacmlApplicationServiceProvider> applicationLoader;
     private static Map<String, XacmlApplicationServiceProvider> providerActionMap = new HashMap<>();
+    private static List<ToscaPolicyTypeIdentifier> toscaPolicyTypeIdents = new ArrayList<>();
 
     private XacmlPdpApplicationManager() {
         super();
@@ -58,19 +60,17 @@ public class XacmlPdpApplicationManager {
         // Load service
         //
         applicationLoader = ServiceLoader.load(XacmlApplicationServiceProvider.class);
+
         //
-        // Iterate through them
+        // Iterate through them the applications for actions and supported policy types
         //
-        Iterator<XacmlApplicationServiceProvider> iterator = applicationLoader.iterator();
-        while (iterator.hasNext()) {
-            //
-            // Get the application
-            //
-            XacmlApplicationServiceProvider application = iterator.next();
+        for (XacmlApplicationServiceProvider application : applicationLoader) {
+
             LOGGER.info("Application {} supports {}", application.applicationName(),
                     application.supportedPolicyTypes());
+
             //
-            // Iterate each application
+            // Iterate through the actions and save in the providerActionMap
             //
             int pathCount = 1;
             for (String action : application.actionDecisionsSupported()) {
@@ -83,7 +83,6 @@ public class XacmlPdpApplicationManager {
                 // May need to scan this name to remove unsafe characters etc.
                 // But for debugging purposes, its good to use the application name
                 //
-                //
                 Path path = Paths.get(applicationPath.toAbsolutePath().toString(),
                         application.applicationName(), Integer.toString(pathCount++));
                 //
@@ -91,6 +90,15 @@ public class XacmlPdpApplicationManager {
                 //
                 application.initialize(path);
             }
+
+            // Get string list of supportedPolicyTypes
+            List<String> supportedPolicyTypes = application.supportedPolicyTypes();
+
+            // Iterate through the supportedPolicyTypes to set the toscaPolicyTypeIdents
+            for (String name : supportedPolicyTypes) {
+                ToscaPolicyTypeIdentifier ident = new ToscaPolicyTypeIdentifier(name, "1.0.0");
+                toscaPolicyTypeIdents.add(ident);
+            }
         }
         //
         // we have initialized
@@ -102,6 +110,10 @@ public class XacmlPdpApplicationManager {
         return providerActionMap.get(request.getAction());
     }
 
+    public static List<ToscaPolicyTypeIdentifier> getToscaPolicyTypeIdents() {
+        return toscaPolicyTypeIdents;
+    }
+
     /**
      * Returns the current count of policy types supported. This could be misleading a bit
      * as some applications can support wildcard of policy types. Eg. onap.Monitoring.* as
@@ -110,10 +122,8 @@ public class XacmlPdpApplicationManager {
      * @return Total count added from all applications
      */
     public static long getPolicyTypeCount() {
-        Iterator<XacmlApplicationServiceProvider> iterator = applicationLoader.iterator();
         long types = 0;
-        while (iterator.hasNext()) {
-            XacmlApplicationServiceProvider application = iterator.next();
+        for (XacmlApplicationServiceProvider application : applicationLoader) {
             types += application.supportedPolicyTypes().size();
         }
         return types;
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpRestServer.java b/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpRestServer.java
index 67bbccaa..f7f6cc38 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpRestServer.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/rest/XacmlPdpRestServer.java
@@ -67,10 +67,12 @@ public class XacmlPdpRestServer implements Startable {
             // Initialize the applications - SEND PROPERTIES
             //
             XacmlPdpApplicationManager.initializeApplications(Paths.get(applicationPath));
+
             //
             // Update statistics manager on the policy types
             //
             XacmlPdpStatisticsManager.setTotalPolicyTypesCount(XacmlPdpApplicationManager.getPolicyTypeCount());
+
             //
             // Get the server properties
             //
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/HealthCheckProvider.java b/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/HealthCheckProvider.java
index 1c77d552..71a4e4e6 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/HealthCheckProvider.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/HealthCheckProvider.java
@@ -43,9 +43,12 @@ public class HealthCheckProvider {
         final HealthCheckReport report = new HealthCheckReport();
         report.setName(NAME);
         report.setUrl(URL);
-        report.setHealthy(XacmlPdpActivator.isAlive());
-        report.setCode(XacmlPdpActivator.isAlive() ? 200 : 500);
-        report.setMessage(XacmlPdpActivator.isAlive() ? ALIVE : NOT_ALIVE);
+
+        boolean alive = XacmlPdpActivator.getCurrent().isAlive();
+
+        report.setHealthy(alive);
+        report.setCode(alive ? 200 : 500);
+        report.setMessage(alive ? ALIVE : NOT_ALIVE);
         return report;
     }
 }
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/StatisticsProvider.java b/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/StatisticsProvider.java
index 1b90f6b8..59d3722c 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/StatisticsProvider.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/rest/provider/StatisticsProvider.java
@@ -37,7 +37,7 @@ public class StatisticsProvider {
      */
     public StatisticsReport fetchCurrentStatistics() {
         final StatisticsReport report = new StatisticsReport();
-        report.setCode(XacmlPdpActivator.isAlive() ? 200 : 500);
+        report.setCode(XacmlPdpActivator.getCurrent().isAlive() ? 200 : 500);
         report.setTotalPolicyTypesCount(XacmlPdpStatisticsManager.getTotalPolicyTypesCount());
         report.setTotalPoliciesCount(XacmlPdpStatisticsManager.getTotalPoliciesCount());
         report.setPermitDecisionsCount(XacmlPdpStatisticsManager.getPermitDecisionsCount());
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/startstop/Main.java b/main/src/main/java/org/onap/policy/pdpx/main/startstop/Main.java
index 65e0a31e..b5915e11 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/startstop/Main.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/startstop/Main.java
@@ -20,8 +20,13 @@
 
 package org.onap.policy.pdpx.main.startstop;
 
+import java.io.FileInputStream;
+import java.net.UnknownHostException;
 import java.util.Arrays;
+import java.util.Properties;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
+import org.onap.policy.pdpx.main.comm.XacmlPdpPapRegistration;
 import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterGroup;
 import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterHandler;
 import org.slf4j.Logger;
@@ -78,13 +83,25 @@ public class Main {
             return;
         }
 
+        // Read the properties
+        Properties props = new Properties();
+        try {
+            String propFile = arguments.getFullPropertyFilePath();
+            try (FileInputStream stream = new FileInputStream(propFile)) {
+                props.load(stream);
+            }
+        } catch (final Exception e) {
+            LOGGER.error("start of xacml pdp service failed", e);
+            return;
+        }
+
         // Now, create the activator for the policy xacml pdp service
-        activator = new XacmlPdpActivator(parameterGroup);
+        activator = new XacmlPdpActivator(parameterGroup, props);
 
         // Start the activator
         try {
-            activator.initialize();
-        } catch (final PolicyXacmlPdpException e) {
+            activator.start();
+        } catch (final RuntimeException e) {
             LOGGER.error("start of policy xacml pdp service failed, used parameters are " + Arrays.toString(args), e);
             return;
         }
@@ -117,13 +134,13 @@ public class Main {
      *
      * @throws PolicyXacmlPdpException on shutdown errors
      */
-    public void shutdown() throws PolicyXacmlPdpException {
+    public void shutdown() {
         // clear the parameterGroup variable
         parameterGroup = null;
 
         // clear the xacml pdp activator
         if (activator != null) {
-            activator.terminate();
+            activator.stop();
         }
     }
 
@@ -139,12 +156,8 @@ public class Main {
          */
         @Override
         public void run() {
-            try {
-                // Shutdown the policy xacml pdp service and wait for everything to stop
-                activator.terminate();
-            } catch (final PolicyXacmlPdpException e) {
-                LOGGER.warn("error occured during shut down of the policy xacml pdp service", e);
-            }
+            // Shutdown the policy xacml pdp service and wait for everything to stop
+            activator.stop();
         }
     }
 
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpActivator.java b/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpActivator.java
index 42ad7bb0..5beeed38 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpActivator.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpActivator.java
@@ -20,8 +20,24 @@
 
 package org.onap.policy.pdpx.main.startstop;
 
+import java.util.Arrays;
+import java.util.Properties;
+import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
+import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
+import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
 import org.onap.policy.common.parameters.ParameterService;
+import org.onap.policy.common.utils.services.ServiceManagerContainer;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.pdp.enums.PdpMessageType;
+import org.onap.policy.models.pdp.enums.PdpState;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
+import org.onap.policy.pdpx.main.PolicyXacmlPdpRuntimeException;
+import org.onap.policy.pdpx.main.comm.XacmlPdpMessage;
+import org.onap.policy.pdpx.main.comm.XacmlPdpPapRegistration;
+import org.onap.policy.pdpx.main.comm.listeners.XacmlPdpStateChangeListener;
 import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterGroup;
 import org.onap.policy.pdpx.main.rest.XacmlPdpRestServer;
 import org.slf4j.Logger;
@@ -31,69 +47,109 @@ import org.slf4j.LoggerFactory;
  * This class wraps a distributor so that it can be activated as a complete service together with
  * all its xacml pdp and forwarding handlers.
  */
-public class XacmlPdpActivator {
+public class XacmlPdpActivator extends ServiceManagerContainer {
+
     // The logger for this class
     private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpActivator.class);
 
+    private static final String[] MSG_TYPE_NAMES = {"messageName"};
+    private static final String TOPIC = "POLICY-PDP-PAP";
+
     // The parameters of this policy xacml pdp activator
     private final XacmlPdpParameterGroup xacmlPdpParameterGroup;
 
-    private static boolean alive = false;
-
+    /**
+     * The XACML PDP REST API server.
+     */
     private XacmlPdpRestServer restServer;
 
     /**
-     * Instantiate the activator for policy xacml pdp as a complete service.
-     *
-     * @param xacmlPdpParameterGroup the parameters for the xacml pdp service
+     * Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then
+     * dispatches them to {@link #pdpUpdateListener}.
      */
-    public XacmlPdpActivator(final XacmlPdpParameterGroup xacmlPdpParameterGroup) {
-        this.xacmlPdpParameterGroup = xacmlPdpParameterGroup;
-    }
+    private final MessageTypeDispatcher msgDispatcher;
 
     /**
-     * Initialize xacml pdp as a complete service.
-     *
-     * @throws PolicyXacmlPdpException on errors in initializing the service
+     * Listens for {@link PdpUpdate} messages and then routes them to the listener associated with the
+     * ID of the originating request.
      */
-    public void initialize() throws PolicyXacmlPdpException {
-        LOGGER.debug("Policy xacml pdp starting as a service . . .");
-        startXacmlPdpRestServer();
-        registerToParameterService(xacmlPdpParameterGroup);
-        XacmlPdpActivator.setAlive(true);
-        LOGGER.debug("Policy xacml pdp started as a service");
-    }
+    private final XacmlPdpStateChangeListener pdpStateChangeListener;
 
     /**
-     * Starts the xacml pdp rest server using configuration parameters.
-     *
-     * @throws PolicyXacmlPdpException if server start fails
+     * The current activator.
      */
-    private void startXacmlPdpRestServer() throws PolicyXacmlPdpException {
-        xacmlPdpParameterGroup.getRestServerParameters().setName(xacmlPdpParameterGroup.getName());
-        restServer = new XacmlPdpRestServer(xacmlPdpParameterGroup.getRestServerParameters(),
-                xacmlPdpParameterGroup.getApplicationPath());
-        if (!restServer.start()) {
-            throw new PolicyXacmlPdpException("Failed to start xacml pdp rest server. Check log for more details...");
-        }
-    }
+    private static XacmlPdpActivator current = null;
+
+    private volatile boolean alive = false;
+
+    private XacmlPdpPapRegistration register;
+
+    private XacmlPdpMessage message;
 
     /**
-     * Terminate policy xacml pdp.
+     * Instantiate the activator for policy xacml pdp as a complete service.
      *
-     * @throws PolicyXacmlPdpException on termination errors
+     * @param xacmlPdpParameterGroup the parameters for the xacml pdp service
+     * @param topicProperties properties used to configure the topics
      */
-    public void terminate() throws PolicyXacmlPdpException {
+    public XacmlPdpActivator(final XacmlPdpParameterGroup xacmlPdpParameterGroup, Properties topicProperties) {
+        TopicEndpoint.manager.addTopicSinks(topicProperties);
+        TopicEndpoint.manager.addTopicSources(topicProperties);
+
         try {
-            deregisterToParameterService(xacmlPdpParameterGroup);
-            XacmlPdpActivator.setAlive(false);
-
-            // Stop the xacml pdp rest server
-            restServer.stop();
-        } catch (final Exception exp) {
-            LOGGER.error("Policy xacml pdp service termination failed", exp);
-            throw new PolicyXacmlPdpException(exp.getMessage(), exp);
+            TopicSinkClient sinkClient = new TopicSinkClient(TOPIC);
+            this.xacmlPdpParameterGroup = xacmlPdpParameterGroup;
+            this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+            this.pdpStateChangeListener = new XacmlPdpStateChangeListener(sinkClient);
+            this.register = new XacmlPdpPapRegistration(sinkClient);
+            this.message = new XacmlPdpMessage();
+        } catch (RuntimeException | TopicSinkClientException e) {
+            throw new PolicyXacmlPdpRuntimeException(e.getMessage(), e);
         }
+
+        xacmlPdpParameterGroup.getRestServerParameters().setName(xacmlPdpParameterGroup.getName());
+
+        // @formatter:off
+        addAction("XACML PDP parameters", () -> ParameterService.register(xacmlPdpParameterGroup),
+            () -> ParameterService.deregister(xacmlPdpParameterGroup.getName()));
+
+        addAction("Request ID Dispatcher",
+            () -> msgDispatcher.register(PdpMessageType.PDP_STATE_CHANGE.name(), this.pdpStateChangeListener),
+            () -> msgDispatcher.unregister(PdpMessageType.PDP_STATE_CHANGE.name()));
+
+        addAction("Message Dispatcher",
+            () -> registerMsgDispatcher(),
+            () -> unregisterMsgDispatcher());
+
+        addAction("topics",
+            () -> TopicEndpoint.manager.start(),
+            () -> TopicEndpoint.manager.shutdown());
+
+        addAction("Create REST server",
+            () -> {
+                restServer = new XacmlPdpRestServer(xacmlPdpParameterGroup.getRestServerParameters(),
+                        xacmlPdpParameterGroup.getApplicationPath());
+            },
+            () -> {
+                restServer = null;
+            });
+
+        addAction("REST server",
+            () -> restServer.start(),
+            () -> restServer.stop());
+
+        addAction("set alive", () -> setAlive(true), () -> setAlive(false));
+
+        addAction("Initial Registration with PAP",
+            () -> {
+                register.pdpRegistration(message.formatStatusMessage(PdpState.PASSIVE));
+            },
+            () -> {
+                register.pdpRegistration(message.formatStatusMessage(PdpState.TERMINATED));
+            });
+        // @formatter:on
+
+        current = this;
     }
 
     /**
@@ -124,11 +180,30 @@ public class XacmlPdpActivator {
     }
 
     /**
+     * Registers the dispatcher with the topic source(s).
+     */
+    private void registerMsgDispatcher() {
+        for (TopicSource source : TopicEndpoint.manager.getTopicSources(Arrays.asList(TOPIC))) {
+            source.register(msgDispatcher);
+        }
+    }
+
+    /**
+     * Unregisters the dispatcher from the topic source(s).
+     */
+    private void unregisterMsgDispatcher() {
+        for (TopicSource source : TopicEndpoint.manager.getTopicSources(Arrays.asList(TOPIC))) {
+            source.unregister(msgDispatcher);
+        }
+    }
+
+    /**
      * Returns the alive status of xacml pdp service.
      *
      * @return the alive
      */
-    public static boolean isAlive() {
+    @Override
+    public boolean isAlive() {
         return alive;
     }
 
@@ -137,7 +212,15 @@ public class XacmlPdpActivator {
      *
      * @param status the status
      */
-    private static void setAlive(final boolean status) {
+    private void setAlive(final boolean status) {
         alive = status;
     }
+
+    public static XacmlPdpActivator getCurrent() {
+        return current;
+    }
+
+    public static void setCurrent(XacmlPdpActivator current) {
+        XacmlPdpActivator.current = current;
+    }
 }
diff --git a/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpCommandLineArguments.java b/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpCommandLineArguments.java
index 1b0f62e3..7a04d237 100644
--- a/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpCommandLineArguments.java
+++ b/main/src/main/java/org/onap/policy/pdpx/main/startstop/XacmlPdpCommandLineArguments.java
@@ -48,6 +48,7 @@ public class XacmlPdpCommandLineArguments {
 
     // The command line options
     private String configurationFilePath = null;
+    private String propertyFilePath = null;
 
     /**
      * Construct the options for the CLI editor.
@@ -76,6 +77,15 @@ public class XacmlPdpCommandLineArguments {
                 .required(false)
                 .type(String.class)
                 .build());
+        options.addOption(Option.builder("p")
+                .longOpt("property-file")
+                .desc("the full path to the topic property file to use, "
+                        + "the property file contains the policy pap topic properties")
+                .hasArg()
+                .argName("PROP_FILE")
+                .required(false)
+                .type(String.class)
+                .build());
         //@formatter:on
     }
 
@@ -106,6 +116,7 @@ public class XacmlPdpCommandLineArguments {
     public String parse(final String[] args) throws PolicyXacmlPdpException {
         // Clear all our arguments
         setConfigurationFilePath(null);
+        setPropertyFilePath(null);
 
         CommandLine commandLine = null;
         try {
@@ -137,6 +148,10 @@ public class XacmlPdpCommandLineArguments {
             setConfigurationFilePath(commandLine.getOptionValue('c'));
         }
 
+        if (commandLine.hasOption('p')) {
+            setPropertyFilePath(commandLine.getOptionValue('p'));
+        }
+
         return null;
     }
 
@@ -213,6 +228,34 @@ public class XacmlPdpCommandLineArguments {
     }
 
     /**
+     * Gets the property file path.
+     *
+     * @return the property file path
+     */
+    public String getPropertyFilePath() {
+        return propertyFilePath;
+    }
+
+    /**
+     * Gets the full expanded property file path.
+     *
+     * @return the property file path
+     */
+    public String getFullPropertyFilePath() {
+        return ResourceUtils.getFilePath4Resource(getPropertyFilePath());
+    }
+
+    /**
+     * Sets the property file path.
+     *
+     * @param propertyFilePath the property file path
+     */
+    public void setPropertyFilePath(final String propertyFilePath) {
+        this.propertyFilePath = propertyFilePath;
+
+    }
+
+    /**
      * Validate readable file.
      *
      * @param fileTag the file tag
diff --git a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestDecision.java b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestDecision.java
index fe307e4f..c93ba6f7 100644
--- a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestDecision.java
+++ b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestDecision.java
@@ -29,7 +29,6 @@ import java.security.NoSuchAlgorithmException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-
 import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
 import javax.ws.rs.client.Entity;
@@ -38,7 +37,6 @@ import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;
-
 import org.glassfish.jersey.client.ClientConfig;
 import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
 import org.junit.AfterClass;
@@ -119,7 +117,8 @@ public class TestDecision {
     }
 
     private static Main startXacmlPdpService() {
-        final String[] XacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json"};
+        final String[] XacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json", "-p",
+            "parameters/topic.properties"};
         return new Main(XacmlPdpConfigParameters);
     }
 
diff --git a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpRestServer.java b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpRestServer.java
index fc3edd73..0deab9d0 100644
--- a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpRestServer.java
+++ b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpRestServer.java
@@ -46,6 +46,7 @@ import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
 import org.onap.policy.common.endpoints.report.HealthCheckReport;
 import org.onap.policy.common.utils.network.NetworkUtil;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
@@ -113,7 +114,7 @@ public class TestXacmlPdpRestServer {
     }
 
     @Test
-    public void testHealthCheckSuccess() throws IOException, InterruptedException {
+    public void testHealthCheckSuccess() throws IOException, InterruptedException, TopicSinkClientException {
         main = startXacmlPdpService(true);
         final Invocation.Builder invocationBuilder = sendHttpRequest(HEALTHCHECK_ENDPOINT);
         final HealthCheckReport report = invocationBuilder.get(HealthCheckReport.class);
@@ -142,7 +143,7 @@ public class TestXacmlPdpRestServer {
     }
 
     @Test
-    public void testStatistics_200() throws IOException, InterruptedException {
+    public void testStatistics_200() throws IOException, InterruptedException, TopicSinkClientException {
         main = startXacmlPdpService(true);
         Invocation.Builder invocationBuilder = sendHttpRequest(STATISTICS_ENDPOINT);
         StatisticsReport report = invocationBuilder.get(StatisticsReport.class);
@@ -188,11 +189,13 @@ public class TestXacmlPdpRestServer {
         }
     }
 
-    private Main startXacmlPdpService(final boolean http) {
-        final String[] xacmlPdpConfigParameters = new String[2];
+    private Main startXacmlPdpService(final boolean http) throws TopicSinkClientException {
+        final String[] xacmlPdpConfigParameters = new String[4];
         if (http) {
             xacmlPdpConfigParameters[0] = "-c";
             xacmlPdpConfigParameters[1] = "parameters/XacmlPdpConfigParameters.json";
+            xacmlPdpConfigParameters[2] = "-p";
+            xacmlPdpConfigParameters[3] = "parameters/topic.properties";
         } else {
             final Properties systemProps = System.getProperties();
             systemProps.put("javax.net.ssl.keyStore", KEYSTORE);
@@ -200,6 +203,8 @@ public class TestXacmlPdpRestServer {
             System.setProperties(systemProps);
             xacmlPdpConfigParameters[0] = "-c";
             xacmlPdpConfigParameters[1] = "parameters/XacmlPdpConfigParameters_Https.json";
+            xacmlPdpConfigParameters[2] = "-p";
+            xacmlPdpConfigParameters[3] = "parameters/topic.properties";
         }
         return new Main(xacmlPdpConfigParameters);
     }
diff --git a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpStatistics.java b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpStatistics.java
index 8d2f7ce4..5b08aa80 100644
--- a/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpStatistics.java
+++ b/main/src/test/java/org/onap/policy/pdpx/main/rest/TestXacmlPdpStatistics.java
@@ -34,11 +34,11 @@ import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.MediaType;
 import org.glassfish.jersey.client.ClientConfig;
 import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
-import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
 import org.onap.policy.common.utils.network.NetworkUtil;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
 import org.onap.policy.pdpx.main.parameters.CommonTestData;
@@ -111,8 +111,9 @@ public class TestXacmlPdpStatistics {
     }
 
 
-    private Main startXacmlPdpService() {
-        final String[] XacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json"};
+    private Main startXacmlPdpService() throws TopicSinkClientException {
+        final String[] XacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json", "-p",
+            "parameters/topic.properties"};
         return new Main(XacmlPdpConfigParameters);
     }
 
diff --git a/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestMain.java b/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestMain.java
index 8c37acb7..80ee95f6 100644
--- a/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestMain.java
+++ b/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestMain.java
@@ -27,6 +27,7 @@ import static org.junit.Assert.assertTrue;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
 import org.onap.policy.pdpx.main.parameters.CommonTestData;
 
@@ -47,7 +48,7 @@ public class TestMain {
     }
 
     @Test
-    public void testMain() throws PolicyXacmlPdpException {
+    public void testMain() throws PolicyXacmlPdpException, TopicSinkClientException {
         final String[] xacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json"};
         final Main main = new Main(xacmlPdpConfigParameters);
         assertTrue(main.getParameters().isValid());
@@ -56,7 +57,7 @@ public class TestMain {
     }
 
     @Test
-    public void testMain_NoArguments() throws PolicyXacmlPdpException {
+    public void testMain_NoArguments() throws PolicyXacmlPdpException, TopicSinkClientException {
         final String[] xacmlPdpConfigParameters = {};
         final Main main = new Main(xacmlPdpConfigParameters);
         assertNull(main.getParameters());
@@ -64,14 +65,14 @@ public class TestMain {
     }
 
     @Test
-    public void testMain_InvalidArguments() {
+    public void testMain_InvalidArguments() throws TopicSinkClientException {
         final String[] xacmlPdpConfigParameters = {"parameters/XacmlPdpConfigParameters.json"};
         final Main main = new Main(xacmlPdpConfigParameters);
         assertNull(main.getParameters());
     }
 
     @Test
-    public void testMain_Help() {
+    public void testMain_Help() throws TopicSinkClientException {
         final String[] xacmlPdpConfigParameters = {"-h"};
         final Main main = new Main(xacmlPdpConfigParameters);
         final String message = "-h,--help                     outputs the usage of this command";
@@ -80,7 +81,7 @@ public class TestMain {
     }
 
     @Test
-    public void testMain_InvalidParameters() {
+    public void testMain_InvalidParameters() throws TopicSinkClientException {
         final String[] xacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters_InvalidName.json"};
         final Main main = new Main(xacmlPdpConfigParameters);
         assertNull(main.getParameters());
diff --git a/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestXacmlPdpActivator.java b/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestXacmlPdpActivator.java
index 7a514f70..51f737a0 100644
--- a/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestXacmlPdpActivator.java
+++ b/main/src/test/java/org/onap/policy/pdpx/main/startstop/TestXacmlPdpActivator.java
@@ -21,12 +21,19 @@
 package org.onap.policy.pdpx.main.startstop;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.net.UnknownHostException;
+import java.util.Properties;
 import org.junit.After;
 import org.junit.BeforeClass;
 
 import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
 import org.onap.policy.pdpx.main.PolicyXacmlPdpException;
 import org.onap.policy.pdpx.main.parameters.CommonTestData;
 import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterGroup;
@@ -42,28 +49,55 @@ public class TestXacmlPdpActivator {
 
     /**
      * Setup the tests.
-     * @throws PolicyXacmlPdpException when Xacml PDP Exceptional condition occurs
      */
     @BeforeClass
-    public static void setup() throws PolicyXacmlPdpException {
-        final String[] xacmlPdpConfigParameters = {"-c", "parameters/XacmlPdpConfigParameters.json"};
-
+    public static void setup() throws Exception {
+        final String[] xacmlPdpConfigParameters =
+            {"-c", "parameters/XacmlPdpConfigParameters.json", "-p", "parameters/topic.properties"};
         final XacmlPdpCommandLineArguments arguments = new XacmlPdpCommandLineArguments(xacmlPdpConfigParameters);
-
         final XacmlPdpParameterGroup parGroup = new XacmlPdpParameterHandler().getParameters(arguments);
 
-        activator = new XacmlPdpActivator(parGroup);
-        activator.initialize();
+        Properties props = new Properties();
+        String propFile = arguments.getFullPropertyFilePath();
+        try (FileInputStream stream = new FileInputStream(propFile)) {
+            props.load(stream);
+        }
+
+        activator = new XacmlPdpActivator(parGroup, props);
     }
 
     @Test
-    public void testXacmlPdpActivator() throws PolicyXacmlPdpException {
+    public void testXacmlPdpActivator() throws PolicyXacmlPdpException, TopicSinkClientException, UnknownHostException {
+        assertFalse(activator.isAlive());
+        activator.start();
+        assertTrue(activator.isAlive());
         assertTrue(activator.getParameterGroup().isValid());
         assertEquals(CommonTestData.PDPX_GROUP_NAME, activator.getParameterGroup().getName());
+
+    }
+
+    @Test
+    public void testGetCurrent_testSetCurrent() {
+        assertSame(activator, XacmlPdpActivator.getCurrent());
+    }
+
+    @Test
+    public void testTerminate() throws Exception {
+        if (!activator.isAlive()) {
+            activator.start();
+        }
+        activator.stop();
+        assertFalse(activator.isAlive());
     }
 
+    /**
+     * Teardown tests.
+     * @throws PolicyXacmlPdpException on termination errors
+     */
     @After
     public void teardown() throws PolicyXacmlPdpException {
-        activator.terminate();
+        if (activator != null && activator.isAlive()) {
+            activator.stop();
+        }
     }
 }
diff --git a/main/src/test/resources/parameters/topic.properties b/main/src/test/resources/parameters/topic.properties
new file mode 100644
index 00000000..11c17dac
--- /dev/null
+++ b/main/src/test/resources/parameters/topic.properties
@@ -0,0 +1,22 @@
+# ============LICENSE_START=======================================================
+# ONAP PAP
+# ================================================================================
+# Copyright (C) 2019 AT&T 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=========================================================
+
+noop.sink.topics=POLICY-PDP-PAP
+noop.sink.topics.POLICY-PDP-PAP.servers=anyserver
+noop.source.topics=POLICY-PDP-PAP
+noop.source.topics.POLICY-PDP-PAP.servers=anyserver