Add PDP heart beat expiration timer

Added heart beat interval to the PDP-UPDATE message sent in response
to a heart beat message received from a PDP.
Added timers to detect missing heart beats and remove the PDP from
the DB - PdpTracker.
Modified current heart beat listener to update PdpTracker when a
heart beat is received.
Allow 3 missed heart beats instead of 2.

Change-Id: I81621fefbe494e0c4d6f0b9767b00b2a9dd398d8
Issue-ID: POLICY-1795
Signed-off-by: jrh3 <jrh3@att.com>

15 files changed, 488 insertions, 17 deletions

diff --git a/main/src/main/java/org/onap/policy/pap/main/PapConstants.java b/main/src/main/java/org/onap/policy/pap/main/PapConstants.java
index 64401b81..3fc36f35 100644
--- a/main/src/main/java/org/onap/policy/pap/main/PapConstants.java
+++ b/main/src/main/java/org/onap/policy/pap/main/PapConstants.java
@@ -30,6 +30,7 @@ public class PapConstants {
     public static final String REG_STATISTICS_MANAGER = "object:manager/statistics";
     public static final String REG_PDP_MODIFY_LOCK = "lock:pdp";
     public static final String REG_PDP_MODIFY_MAP = "object:pdp/modify/map";
+    public static final String REG_PDP_TRACKER = "object:pdp/tracker";
     public static final String REG_PAP_DAO_FACTORY = "object:pap/dao/factory";
 
     // topic names
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/PdpStatusMessageHandler.java b/main/src/main/java/org/onap/policy/pap/main/comm/PdpStatusMessageHandler.java
index 7ef9c594..f5184c93 100644
--- a/main/src/main/java/org/onap/policy/pap/main/comm/PdpStatusMessageHandler.java
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/PdpStatusMessageHandler.java
@@ -28,6 +28,7 @@ import java.util.Optional;
 import java.util.TreeMap;
 
 import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.onap.policy.common.parameters.ParameterService;
 import org.onap.policy.common.utils.services.Registry;
 import org.onap.policy.models.base.PfModelException;
 import org.onap.policy.models.pdp.concepts.Pdp;
@@ -44,6 +45,7 @@ import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
 import org.onap.policy.pap.main.PapConstants;
 import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
 import org.onap.policy.pap.main.PolicyPapException;
+import org.onap.policy.pap.main.parameters.PapParameterGroup;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,9 +56,10 @@ import org.slf4j.LoggerFactory;
  * @author Ram Krishna Verma (ram.krishna.verma@est.tech)
  */
 public class PdpStatusMessageHandler {
-
     private static final Logger LOGGER = LoggerFactory.getLogger(PdpStatusMessageHandler.class);
 
+    private static final String PAP_GROUP_PARAMS_NAME = "PapGroup";
+
     /**
      * Lock used when updating PDPs.
      */
@@ -70,7 +73,12 @@ public class PdpStatusMessageHandler {
     /**
      * Factory for PAP DAO.
      */
-    PolicyModelsProviderFactoryWrapper modelProviderWrapper;
+    private final PolicyModelsProviderFactoryWrapper modelProviderWrapper;
+
+    /**
+     * Heart beat interval, in milliseconds, to pass to PDPs.
+     */
+    private final long heartBeatMs;
 
     /**
      * Constructs the object.
@@ -79,6 +87,9 @@ public class PdpStatusMessageHandler {
         modelProviderWrapper = Registry.get(PapConstants.REG_PAP_DAO_FACTORY, PolicyModelsProviderFactoryWrapper.class);
         updateLock = Registry.get(PapConstants.REG_PDP_MODIFY_LOCK, Object.class);
         requestMap = Registry.get(PapConstants.REG_PDP_MODIFY_MAP, PdpModifyRequestMap.class);
+
+        PapParameterGroup params = ParameterService.get(PAP_GROUP_PARAMS_NAME);
+        heartBeatMs = params.getPdpParameters().getHeartBeatMs();
     }
 
     /**
@@ -94,6 +105,15 @@ public class PdpStatusMessageHandler {
                 } else {
                     handlePdpHeartbeat(message, databaseProvider);
                 }
+
+                /*
+                 * Indicate that a heart beat was received from the PDP. This is invoked
+                 * only if handleXxx() does not throw an exception.
+                 */
+                if (message.getName() != null) {
+                    PdpTracker pdpTracker = Registry.get(PapConstants.REG_PDP_TRACKER);
+                    pdpTracker.add(message.getName());
+                }
             } catch (final PolicyPapException exp) {
                 LOGGER.error("Operation Failed", exp);
             } catch (final Exception exp) {
@@ -297,6 +317,7 @@ public class PdpStatusMessageHandler {
         update.setPdpGroup(pdpGroupName);
         update.setPdpSubgroup(subGroup.getPdpType());
         update.setPolicies(getToscaPolicies(subGroup, databaseProvider));
+        update.setPdpHeartbeatIntervalMs(heartBeatMs);
 
         LOGGER.debug("Created PdpUpdate message - {}", update);
         return update;
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/PdpTracker.java b/main/src/main/java/org/onap/policy/pap/main/comm/PdpTracker.java
new file mode 100644
index 00000000..04dfe813
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/PdpTracker.java
@@ -0,0 +1,153 @@
+/*
+ * ============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=========================================================
+ */
+
+package org.onap.policy.pap.main.comm;
+
+import java.util.HashMap;
+import java.util.Map;
+import lombok.Builder;
+import lombok.NonNull;
+import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.pdp.concepts.Pdp;
+import org.onap.policy.models.pdp.concepts.PdpGroup;
+import org.onap.policy.models.pdp.concepts.PdpSubGroup;
+import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
+import org.onap.policy.pap.main.PolicyPapRuntimeException;
+import org.onap.policy.pap.main.comm.TimerManager.Timer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tracks PDPs. When a PDP is added to the tracker, a timer is started. If the PDP is not
+ * re-added to the tracker before the timer expires, then
+ * {@link PdpModifyRequestMap#removeFromGroups(String)} is called.
+ */
+public class PdpTracker {
+    private static final Logger logger = LoggerFactory.getLogger(PdpTracker.class);
+
+    /**
+     * PDP expiration timers.
+     */
+    private final TimerManager timers;
+
+    /**
+     * Maps a PDP name to its expiration timer.
+     */
+    private final Map<String, TimerManager.Timer> pdp2timer = new HashMap<>();
+
+    /**
+     * PDP modification lock.
+     */
+    private final Object modifyLock;
+
+    /**
+     * Used to remove a PDP from its group/subgroup.
+     */
+    private final PdpModifyRequestMap requestMap;
+
+
+    /**
+     * Constructs the object. Loads the list of PDPs to be tracked, from the DB.
+     *
+     * @param requestMap map used to remove a PDP from its group/subgroup
+     * @param modifyLock object to be locked while data structures are updated
+     * @param timers timers used to detect missed heart beats
+     * @param daoFactory DAO factory
+     */
+    @Builder
+    public PdpTracker(@NonNull PdpModifyRequestMap requestMap, @NonNull Object modifyLock, @NonNull TimerManager timers,
+                    @NonNull PolicyModelsProviderFactoryWrapper daoFactory) {
+
+        this.requestMap = requestMap;
+        this.modifyLock = modifyLock;
+        this.timers = timers;
+
+        loadPdps(daoFactory);
+    }
+
+    /**
+     * Loads the PDPs from the DB.
+     *
+     * @param daoFactory DAO factory
+     */
+    private void loadPdps(PolicyModelsProviderFactoryWrapper daoFactory) {
+        synchronized (modifyLock) {
+            try (PolicyModelsProvider dao = daoFactory.create()) {
+                for (PdpGroup group : dao.getPdpGroups(null)) {
+                    loadPdpsFromGroup(group);
+                }
+
+            } catch (PfModelException e) {
+                throw new PolicyPapRuntimeException("cannot load PDPs from the DB", e);
+            }
+        }
+    }
+
+    /**
+     * Loads the PDPs appearing within a group.
+     *
+     * @param group group whose PDPs are to be loaded
+     */
+    private void loadPdpsFromGroup(PdpGroup group) {
+        for (PdpSubGroup subgrp : group.getPdpSubgroups()) {
+            for (Pdp pdp : subgrp.getPdpInstances()) {
+                add(pdp.getInstanceId());
+            }
+        }
+    }
+
+    /**
+     * Adds a PDP to the tracker and starts its timer. If a timer is already running, the
+     * old timer is cancelled.
+     *
+     * @param pdpName name of the PDP
+     */
+    public void add(String pdpName) {
+        synchronized (modifyLock) {
+            Timer timer = pdp2timer.remove(pdpName);
+            if (timer != null) {
+                timer.cancel();
+            }
+
+            timer = timers.register(pdpName, this::handleTimeout);
+            pdp2timer.put(pdpName, timer);
+        }
+    }
+
+    /**
+     * Handles a timeout. Removes the PDP from {@link #pdp2timer}.
+     *
+     * @param pdpName name of the PDP whose timer has expired
+     */
+    private void handleTimeout(String pdpName) {
+        synchronized (modifyLock) {
+            // remove timer - no need to cancel it, as TimerManager does that
+            pdp2timer.remove(pdpName);
+
+            try {
+                requestMap.removeFromGroups(pdpName);
+
+            } catch (PfModelException e) {
+                logger.warn("unable to remove PDP {} from its group/subgroup", pdpName, e);
+            }
+        }
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pap/main/parameters/PdpParameters.java b/main/src/main/java/org/onap/policy/pap/main/parameters/PdpParameters.java
index 84fe353b..1776772a 100644
--- a/main/src/main/java/org/onap/policy/pap/main/parameters/PdpParameters.java
+++ b/main/src/main/java/org/onap/policy/pap/main/parameters/PdpParameters.java
@@ -22,6 +22,7 @@ package org.onap.policy.pap.main.parameters;
 
 import lombok.Getter;
 import org.onap.policy.common.parameters.ParameterGroupImpl;
+import org.onap.policy.common.parameters.annotations.Min;
 import org.onap.policy.common.parameters.annotations.NotBlank;
 import org.onap.policy.common.parameters.annotations.NotNull;
 
@@ -32,6 +33,10 @@ import org.onap.policy.common.parameters.annotations.NotNull;
 @NotBlank
 @Getter
 public class PdpParameters extends ParameterGroupImpl {
+
+    @Min(1)
+    private long heartBeatMs;
+
     private PdpUpdateParameters updateParameters;
     private PdpStateChangeParameters stateChangeParameters;
 
diff --git a/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java b/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java
index 1b7281ca..e1ad80e2 100644
--- a/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java
+++ b/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java
@@ -24,7 +24,6 @@ package org.onap.policy.pap.main.startstop;
 import java.util.Arrays;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicReference;
-
 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
 import org.onap.policy.common.endpoints.event.comm.TopicSource;
 import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
@@ -39,6 +38,7 @@ import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
 import org.onap.policy.pap.main.PolicyPapRuntimeException;
 import org.onap.policy.pap.main.comm.PdpHeartbeatListener;
 import org.onap.policy.pap.main.comm.PdpModifyRequestMap;
+import org.onap.policy.pap.main.comm.PdpTracker;
 import org.onap.policy.pap.main.comm.Publisher;
 import org.onap.policy.pap.main.comm.TimerManager;
 import org.onap.policy.pap.main.parameters.PapParameterGroup;
@@ -57,12 +57,12 @@ public class PapActivator extends ServiceManagerContainer {
     private static final String[] MSG_TYPE_NAMES = { "messageName" };
     private static final String[] REQ_ID_NAMES = { "response", "responseTo" };
 
-    private final PapParameterGroup papParameterGroup;
-
     /**
-     * The PAP REST API server.
+     * Max number of heat beats that can be missed before PAP removes a PDP.
      */
-    private PapRestServer restServer;
+    private static final int MAX_MISSED_HEARTBEATS = 3;
+
+    private final PapParameterGroup papParameterGroup;
 
     /**
      * Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then dispatches them to
@@ -110,7 +110,10 @@ public class PapActivator extends ServiceManagerContainer {
         final AtomicReference<Publisher> pdpPub = new AtomicReference<>();
         final AtomicReference<TimerManager> pdpUpdTimers = new AtomicReference<>();
         final AtomicReference<TimerManager> pdpStChgTimers = new AtomicReference<>();
+        final AtomicReference<TimerManager> heartBeatTimers = new AtomicReference<>();
         final AtomicReference<PolicyModelsProviderFactoryWrapper> daoFactory = new AtomicReference<>();
+        final AtomicReference<PdpModifyRequestMap> requestMap = new AtomicReference<>();
+        final AtomicReference<PapRestServer> restServer = new AtomicReference<>();
 
         // @formatter:off
         addAction("PAP parameters",
@@ -153,6 +156,14 @@ public class PapActivator extends ServiceManagerContainer {
             },
             () -> pdpPub.get().stop());
 
+        addAction("PDP heart beat timers",
+            () -> {
+                long maxWaitHeartBeatMs = MAX_MISSED_HEARTBEATS * pdpParams.getHeartBeatMs();
+                heartBeatTimers.set(new TimerManager("heart beat", maxWaitHeartBeatMs));
+                startThread(heartBeatTimers.get());
+            },
+            () -> heartBeatTimers.get().stop());
+
         addAction("PDP update timers",
             () -> {
                 pdpUpdTimers.set(new TimerManager("update", pdpParams.getUpdateParameters().getMaxWaitMs()));
@@ -172,7 +183,8 @@ public class PapActivator extends ServiceManagerContainer {
             () -> Registry.unregister(PapConstants.REG_PDP_MODIFY_LOCK));
 
         addAction("PDP modification requests",
-            () -> Registry.register(PapConstants.REG_PDP_MODIFY_MAP, new PdpModifyRequestMap(
+            () -> {
+                requestMap.set(new PdpModifyRequestMap(
                             new PdpModifyRequestMapParams()
                                     .setDaoFactory(daoFactory.get())
                                     .setModifyLock(pdpUpdateLock)
@@ -180,16 +192,26 @@ public class PapActivator extends ServiceManagerContainer {
                                     .setPublisher(pdpPub.get())
                                     .setResponseDispatcher(reqIdDispatcher)
                                     .setStateChangeTimers(pdpStChgTimers.get())
-                                    .setUpdateTimers(pdpUpdTimers.get()))),
+                                    .setUpdateTimers(pdpUpdTimers.get())));
+                Registry.register(PapConstants.REG_PDP_MODIFY_MAP, requestMap.get());
+            },
             () -> Registry.unregister(PapConstants.REG_PDP_MODIFY_MAP));
 
-        addAction("Create REST server",
-            () -> restServer = new PapRestServer(papParameterGroup.getRestServerParameters()),
-            () -> restServer = null);
+        addAction("PDP heart beat tracker",
+            () -> Registry.register(PapConstants.REG_PDP_TRACKER, PdpTracker.builder()
+                                    .daoFactory(daoFactory.get())
+                                    .timers(heartBeatTimers.get())
+                                    .modifyLock(pdpUpdateLock)
+                                    .requestMap(requestMap.get())
+                                    .build()),
+            () -> Registry.unregister(PapConstants.REG_PDP_TRACKER));
 
         addAction("REST server",
-            () -> restServer.start(),
-            () -> restServer.stop());
+            () -> {
+                restServer.set(new PapRestServer(papParameterGroup.getRestServerParameters()));
+                restServer.get().start();
+            },
+            () -> restServer.get().stop());
         // @formatter:on
     }
 
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/PdpTrackerTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/PdpTrackerTest.java
new file mode 100644
index 00000000..19684a71
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/PdpTrackerTest.java
@@ -0,0 +1,212 @@
+/*-
+ * ============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=========================================================
+ */
+
+package org.onap.policy.pap.main.comm;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.onap.policy.common.utils.resources.ResourceUtils;
+import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.pdp.concepts.PdpGroup;
+import org.onap.policy.models.pdp.concepts.PdpGroups;
+import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
+import org.onap.policy.pap.main.PolicyPapRuntimeException;
+import org.onap.policy.pap.main.comm.TimerManager.Timer;
+
+public class PdpTrackerTest {
+    private static final String PDP1 = "pdp1";
+    private static final String PDP2 = "pdp2";
+
+    private PdpTracker tracker;
+    private PdpTracker.PdpTrackerBuilder builder;
+
+    private Object modifyLock;
+
+    @Captor
+    private ArgumentCaptor<Consumer<String>> handlerCaptor;
+
+    @Mock
+    private PdpModifyRequestMap requestMap;
+
+    @Mock
+    private TimerManager timers;
+
+    @Mock
+    private PolicyModelsProviderFactoryWrapper daoFactory;
+
+    @Mock
+    private PolicyModelsProvider dao;
+
+    @Mock
+    private Timer timer1;
+
+    @Mock
+    private Timer timer2;
+
+    /**
+     * Sets up.
+     *
+     * @throws Exception if an error occurs
+     */
+    @Before
+    public void setUp() throws Exception {
+        MockitoAnnotations.initMocks(this);
+
+        modifyLock = new Object();
+
+        builder = PdpTracker.builder().daoFactory(daoFactory).modifyLock(modifyLock).requestMap(requestMap)
+                        .timers(timers);
+
+        when(daoFactory.create()).thenReturn(dao);
+
+        when(dao.getPdpGroups(null)).thenReturn(Collections.emptyList());
+
+        when(timers.register(eq(PDP1), any())).thenReturn(timer1);
+        when(timers.register(eq(PDP2), any())).thenReturn(timer2);
+
+        tracker = builder.build();
+    }
+
+    @Test
+    public void testPdpTracker() throws Exception {
+        // verify that PDPs were loaded
+        verify(dao).getPdpGroups(null);
+    }
+
+    @Test
+    public void testBuilderToString() throws Exception {
+        assertNotNull(builder.toString());
+    }
+
+    @Test
+    public void testPdpTracker_MissingRequestMap() throws Exception {
+        assertThatThrownBy(() -> builder.requestMap(null).build()).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    public void testPdpTracker_MissingModifyLock() throws Exception {
+        assertThatThrownBy(() -> builder.modifyLock(null).build()).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    public void testPdpTracker_MissingTimers() throws Exception {
+        assertThatThrownBy(() -> builder.timers(null).build()).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    public void testPdpTracker_MissingDaoFactory() throws Exception {
+        assertThatThrownBy(() -> builder.daoFactory(null).build()).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    public void testLoadPdps_testLoadPdpsFromGroup() throws Exception {
+        // arrange for DAO to return a couple of groups
+        String groupsJson = ResourceUtils.getResourceAsString("comm/PdpTracker.json");
+        List<PdpGroup> groups = new StandardCoder().decode(groupsJson, PdpGroups.class).getGroups();
+        when(dao.getPdpGroups(null)).thenReturn(groups);
+
+        tracker = builder.build();
+
+        // verify that all PDPs were registered
+        verify(timers).register(eq("pdp-A"), any());
+        verify(timers).register(eq("pdp-B"), any());
+        verify(timers).register(eq("pdp-C"), any());
+        verify(timers).register(eq("pdp-D"), any());
+    }
+
+    @Test
+    public void testLoadPdps_DaoException() throws Exception {
+        // arrange for DAO to throw an exception
+        PfModelException ex = mock(PfModelException.class);
+        when(daoFactory.create()).thenThrow(ex);
+
+        assertThatThrownBy(() -> builder.build()).isInstanceOf(PolicyPapRuntimeException.class).hasCause(ex);
+    }
+
+    @Test
+    public void testAdd() {
+        tracker.add(PDP1);
+        verify(timers).register(eq(PDP1), any());
+        verify(timer1, never()).cancel();
+
+        tracker.add(PDP2);
+        verify(timers).register(eq(PDP2), any());
+        verify(timer1, never()).cancel();
+        verify(timer2, never()).cancel();
+
+        // re-add PDP1 - old timer should be canceled and a new timer added
+        Timer timer3 = mock(Timer.class);
+        when(timers.register(eq(PDP1), any())).thenReturn(timer3);
+        tracker.add(PDP1);
+        verify(timer1).cancel();
+        verify(timer2, never()).cancel();
+        verify(timer3, never()).cancel();
+    }
+
+    @Test
+    public void testHandleTimeout() throws Exception {
+        tracker.add(PDP1);
+        tracker.add(PDP2);
+
+        verify(timers).register(eq(PDP1), handlerCaptor.capture());
+
+        handlerCaptor.getValue().accept(PDP1);
+
+        verify(requestMap).removeFromGroups(PDP1);
+
+        // now we'll re-add PDP1 - the original timer should not be canceled
+        Timer timer3 = mock(Timer.class);
+        when(timers.register(eq(PDP1), any())).thenReturn(timer3);
+        tracker.add(PDP1);
+        verify(timer1, never()).cancel();
+    }
+
+    @Test
+    public void testHandleTimeout_MapException() throws Exception {
+        tracker.add(PDP1);
+
+        verify(timers).register(eq(PDP1), handlerCaptor.capture());
+
+        // arrange for request map to throw an exception
+        PfModelException ex = mock(PfModelException.class);
+        when(requestMap.removeFromGroups(PDP1)).thenThrow(ex);
+
+        // exception should be caught, but not re-thrown
+        handlerCaptor.getValue().accept(PDP1);
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpParameters.java b/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpParameters.java
index eb9a6e8b..1474bbfb 100644
--- a/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpParameters.java
+++ b/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpParameters.java
@@ -46,6 +46,8 @@ public class TestPdpParameters {
         PdpStateChangeParameters state = params.getStateChangeParameters();
         assertNotNull(state);
         assertEquals(5, state.getMaxWaitMs());
+
+        assertEquals(6L, params.getHeartBeatMs());
     }
 
     @Test
@@ -58,6 +60,13 @@ public class TestPdpParameters {
         assertNull(result.getResult());
         assertTrue(result.isValid());
 
+        // invalid heart beat
+        json2 = json.replaceFirst(": 6", ": 0");
+        result = coder.decode(json2, PapParameterGroup.class).getPdpParameters().validate();
+        assertFalse(result.isValid());
+        assertTrue(result.getResult().contains(
+                        "field 'heartBeatMs' type 'long' value '0' INVALID, must be >= 1".replace('\'', '"')));
+
         // no update params
         json2 = testData.nullifyField(json, "updateParameters");
         result = coder.decode(json2, PapParameterGroup.class).getPdpParameters().validate();
@@ -66,7 +75,7 @@ public class TestPdpParameters {
         assertTrue(result.getResult().contains("is null"));
 
         // invalid update params
-        json2 = json.replaceFirst("2", "-2");
+        json2 = json.replaceFirst(": 2", ": -2");
         result = coder.decode(json2, PapParameterGroup.class).getPdpParameters().validate();
         assertFalse(result.isValid());
         assertTrue(result.getResult().contains("parameter group 'PdpUpdateParameters'".replace('\'', '"')));
@@ -79,7 +88,7 @@ public class TestPdpParameters {
         assertFalse(result.isValid());
 
         // invalid state-change params
-        json2 = json.replaceFirst("5", "-5");
+        json2 = json.replaceFirst(": 5", ": -5");
         result = coder.decode(json2, PapParameterGroup.class).getPdpParameters().validate();
         assertFalse(result.isValid());
         assertTrue(result.getResult().contains("parameter group 'PdpStateChangeParameters'".replace('\'', '"')));
diff --git a/main/src/test/resources/comm/PdpTracker.json b/main/src/test/resources/comm/PdpTracker.json
new file mode 100644
index 00000000..5c8942d5
--- /dev/null
+++ b/main/src/test/resources/comm/PdpTracker.json
@@ -0,0 +1,41 @@
+{
+    "groups": [
+        {
+            "name": "group-X",
+            "pdpSubgroups": [
+                {
+                    "pdpType": "apex",
+                    "pdpInstances": [
+                        {
+                            "instanceId": "pdp-A"
+                        },
+                        {
+                            "instanceId": "pdp-B"
+                        }
+                    ]
+                },
+                {
+                    "pdpType": "drools",
+                    "pdpInstances": [
+                        {
+                            "instanceId": "pdp-C"
+                        }
+                    ]
+                }
+            ]
+        },
+        {
+            "name": "group-Y",
+            "pdpSubgroups": [
+                {
+                    "pdpType": "apex",
+                    "pdpInstances": [
+                        {
+                            "instanceId": "pdp-D"
+                        }
+                    ]
+                }
+            ]
+        } 
+    ]
+}
diff --git a/main/src/test/resources/e2e/PapConfigParameters.json b/main/src/test/resources/e2e/PapConfigParameters.json
index 4e085f76..871077c5 100644
--- a/main/src/test/resources/e2e/PapConfigParameters.json
+++ b/main/src/test/resources/e2e/PapConfigParameters.json
@@ -8,6 +8,7 @@
         "https": true
     },
     "pdpParameters": {
+        "heartBeatMs": 5000,
         "updateParameters": {
             "maxRetryCount": 0,
             "maxWaitMs": 5000
diff --git a/main/src/test/resources/parameters/MinimumParameters.json b/main/src/test/resources/parameters/MinimumParameters.json
index e611dc90..99c8a3a3 100644
--- a/main/src/test/resources/parameters/MinimumParameters.json
+++ b/main/src/test/resources/parameters/MinimumParameters.json
@@ -7,6 +7,7 @@
         "password":"zb!XztG34"
     },
     "pdpParameters": {
+        "heartBeatMs": 1,
         "updateParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 1
diff --git a/main/src/test/resources/parameters/PapConfigParameters.json b/main/src/test/resources/parameters/PapConfigParameters.json
index 2c24e4fe..16b22ae5 100644
--- a/main/src/test/resources/parameters/PapConfigParameters.json
+++ b/main/src/test/resources/parameters/PapConfigParameters.json
@@ -8,6 +8,7 @@
         "https": true
     },
     "pdpParameters": {
+        "heartBeatMs": 10,
         "updateParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 1
diff --git a/main/src/test/resources/parameters/PapConfigParametersStd.json b/main/src/test/resources/parameters/PapConfigParametersStd.json
index 6089c0b7..853249e0 100644
--- a/main/src/test/resources/parameters/PapConfigParametersStd.json
+++ b/main/src/test/resources/parameters/PapConfigParametersStd.json
@@ -15,7 +15,8 @@
         "stateChangeParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 5
-        }
+        },
+        "heartBeatMs": 6
     },
     "databaseProviderParameters": {
         "name": "PolicyModelsProviderParameters",
diff --git a/main/src/test/resources/parameters/PapConfigParameters_InvalidName.json b/main/src/test/resources/parameters/PapConfigParameters_InvalidName.json
index d06ecfc8..15e608ba 100644
--- a/main/src/test/resources/parameters/PapConfigParameters_InvalidName.json
+++ b/main/src/test/resources/parameters/PapConfigParameters_InvalidName.json
@@ -7,6 +7,7 @@
         "password":"zb!XztG34"
     },
     "pdpParameters": {
+        "heartBeatMs": 1,
         "updateParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 1
diff --git a/main/src/test/resources/parameters/PapConfigParameters_sim.json b/main/src/test/resources/parameters/PapConfigParameters_sim.json
index a9a68f90..96bbedd0 100644
--- a/main/src/test/resources/parameters/PapConfigParameters_sim.json
+++ b/main/src/test/resources/parameters/PapConfigParameters_sim.json
@@ -8,6 +8,7 @@
         "https": true
     },
     "pdpParameters": {
+        "heartBeatMs": 10,
         "updateParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 30000
diff --git a/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json b/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json
index 32b6c30f..519060dd 100644
--- a/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json
+++ b/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json
@@ -9,6 +9,7 @@
         "aaf": false
     },
     "pdpParameters": {
+        "heartBeatMs": 120000,
         "updateParameters": {
             "maxRetryCount": 1,
             "maxWaitMs": 30000