Add comprehensive junit tests for pooling feature

Added FeatureTest.java and FeatureTest2.java.
Fixed a number of issues that were identified via the above
tests.
Moved fixes of policy-endpoints to other JIRA tickets.
Resolved potential sonar issue with commented-out code by
making STD_XXX constants into plain variables to they could be
changed via a method call.
Changed stdEventWaitMs to stdEventWaitSec.

Change-Id: Icf8fcc9b0dfe2578aa0787e0c9224a6f76a068ee
Issue-ID: POLICY-748
Signed-off-by: Jim Hahn <jrh3@att.com>

12 files changed, 165 insertions, 129 deletions

diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
index 102eda75..abb4da33 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/DmaapManager.java
@@ -202,6 +202,7 @@ public class DmaapManager {
 
         } catch (InterruptedException e) {
             logger.warn("message transmission stopped due to {}", e.getMessage());
+            Thread.currentThread().interrupt();
         }
 
         try {
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
index 21cbc4db..2bec4579 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingFeature.java
@@ -21,6 +21,7 @@
 package org.onap.policy.drools.pooling;
 
 import java.util.Properties;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 import org.onap.policy.common.utils.properties.exception.PropertyException;
@@ -54,6 +55,11 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
     private static Factory factory;
 
     /**
+     * ID of this host.
+     */
+    private final String host;
+
+    /**
      * Entire set of feature properties, including those specific to various controllers.
      */
     private Properties featProps = null;
@@ -75,6 +81,8 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
      */
     public PoolingFeature() {
         super();
+
+        this.host = UUID.randomUUID().toString();
     }
 
     protected static Factory getFactory() {
@@ -90,6 +98,10 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
         PoolingFeature.factory = factory;
     }
 
+    public String getHost() {
+        return host;
+    }
+
     @Override
     public int getSequenceNumber() {
         return 0;
@@ -123,7 +135,7 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
                 PoolingProperties props = new PoolingProperties(name, featProps);
 
                 logger.info("pooling enabled for {}", name);
-                ctlr2pool.computeIfAbsent(name, xxx -> factory.makeManager(controller, props));
+                ctlr2pool.computeIfAbsent(name, xxx -> factory.makeManager(host, controller, props));
 
             } catch (PropertyException e) {
                 logger.error("pooling disabled due to exception for {}", name, e);
@@ -371,12 +383,13 @@ public class PoolingFeature implements PolicyEngineFeatureAPI, PolicyControllerF
         /**
          * Makes a pooling manager for a controller.
          * 
+         * @param host name/uuid of this host
          * @param controller
          * @param props properties to use to configure the manager
          * @return a new pooling manager
          */
-        public PoolingManagerImpl makeManager(PolicyController controller, PoolingProperties props) {
-            return new PoolingManagerImpl(controller, props);
+        public PoolingManagerImpl makeManager(String host, PolicyController controller, PoolingProperties props) {
+            return new PoolingManagerImpl(host, controller, props);
         }
 
         /**
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
index 422efdd7..d2312469 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingManagerImpl.java
@@ -25,12 +25,10 @@ import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Map;
 import java.util.Properties;
-import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import org.onap.policy.drools.controller.DroolsController;
 import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.drools.event.comm.TopicListener;
@@ -47,6 +45,7 @@ import org.onap.policy.drools.pooling.state.QueryState;
 import org.onap.policy.drools.pooling.state.StartState;
 import org.onap.policy.drools.pooling.state.State;
 import org.onap.policy.drools.pooling.state.StateTimerTask;
+import org.onap.policy.drools.properties.PolicyProperties;
 import org.onap.policy.drools.protocol.coders.EventProtocolCoder;
 import org.onap.policy.drools.system.PolicyController;
 import org.slf4j.Logger;
@@ -75,11 +74,6 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
     private static Factory factory = new Factory();
 
     /**
-     * ID of the last host that was created.
-     */
-    private static final AtomicReference<String> lastHost = new AtomicReference<>(null);
-
-    /**
      * ID of this host.
      */
     private final String host;
@@ -158,28 +152,24 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
 
     /**
      * Constructs the manager, initializing all of the data structures.
-     * 
+     *
+     * @param host name/uuid of this host
      * @param controller controller with which this is associated
      * @param props feature properties specific to the controller
      */
-    public PoolingManagerImpl(PolicyController controller, PoolingProperties props) {
-        this.host = UUID.randomUUID().toString();
+    public PoolingManagerImpl(String host, PolicyController controller, PoolingProperties props) {
+        this.host = host;
         this.controller = controller;
         this.props = props;
 
-        lastHost.set(this.host);
-
         try {
             this.listener = (TopicListener) controller;
             this.serializer = new Serializer();
             this.topic = props.getPoolingTopic();
             this.eventq = factory.makeEventQueue(props);
-
-            SpecProperties spec = new SpecProperties(PoolingProperties.PROP_EXTRACTOR_PREFIX, controller.getName(),
-                            props.getSource());
-            this.extractors = factory.makeClassExtractors(spec);
-
-            this.dmaapMgr = factory.makeDmaapManager(props);
+            this.extractors = factory.makeClassExtractors(makeExtractorProps(controller, props.getSource()));
+            this.dmaapMgr = factory.makeDmaapManager(props.getPoolingTopic(),
+                            makeDmaapProps(controller, props.getSource()));
             this.current = new IdleState(this);
 
             logger.info("allocating host {} to controller {} for topic {}", host, controller.getName(), topic);
@@ -203,16 +193,6 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
     }
 
     /**
-     * Used by junit tests.
-     * 
-     * @return the ID of the last host that was created, or {@code null} if no hosts have
-     *         been created yet
-     */
-    protected static String getLastHost() {
-        return lastHost.get();
-    }
-
-    /**
      * Should only be used by junit tests.
      * 
      * @return the current state
@@ -237,6 +217,50 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
     }
 
     /**
+     * Makes properties for configuring extractors.
+     * 
+     * @param controller the controller for which the extractors will be configured
+     * @param source properties from which to get the extractor properties
+     * @return extractor properties
+     */
+    private Properties makeExtractorProps(PolicyController controller, Properties source) {
+        return new SpecProperties(PoolingProperties.PROP_EXTRACTOR_PREFIX, controller.getName(), source);
+    }
+
+    /**
+     * Makes properties for configuring DMaaP. Copies properties from the source that
+     * start with the Pooling property prefix followed by the controller name, stripping
+     * the prefix and controller name.
+     * 
+     * @param controller the controller for which DMaaP will be configured
+     * @param source properties from which to get the DMaaP properties
+     * @return DMaaP properties
+     */
+    private Properties makeDmaapProps(PolicyController controller, Properties source) {
+        SpecProperties props = new SpecProperties("", "controller." + controller.getName(), source);
+
+        // could be UEB or DMAAP, so add both
+        addDmaapConsumerProps(props, PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS);
+        addDmaapConsumerProps(props, PolicyProperties.PROPERTY_DMAAP_SOURCE_TOPICS);
+
+        return props;
+    }
+
+    /**
+     * Adds DMaaP consumer properties, consumer group & instance. The group is the host
+     * and the instance is a constant.
+     * 
+     * @param props where to add the new properties
+     * @param prefix property prefix
+     */
+    private void addDmaapConsumerProps(SpecProperties props, String prefix) {
+        String fullpfx = props.getSpecPrefix() + prefix + "." + topic;
+        
+        props.setProperty(fullpfx + PolicyProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX, host);
+        props.setProperty(fullpfx + PolicyProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX, "0");
+    }
+
+    /**
      * Indicates that the controller is about to start. Starts the publisher for the
      * internal topic, and creates a thread pool for the timers.
      * 
@@ -288,8 +312,8 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
             scheduler = null;
 
             if (!(current instanceof IdleState)) {
-                dmaapMgr.stopConsumer(this);
                 changeState(new IdleState(this));
+                dmaapMgr.stopConsumer(this);
                 publishAdmin(new Offline(getHost()));
             }
         }
@@ -751,26 +775,27 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
 
     @Override
     public CountDownLatch startDistributing(BucketAssignments asgn) {
-        if (asgn == null) {
-            return null;
-        }
-
-        logger.info("new assignments for topic {}", getTopic());
-
         synchronized (curLocker) {
+            int sz = (asgn == null ? 0 : asgn.getAllHosts().size());
+            logger.info("new assignments for {} hosts on topic {}", sz, getTopic());
             assignments = asgn;
         }
 
+        if (asgn == null) {
+            return null;
+        }
+
         /*
          * publish the events from the event queue, but do it in a background thread so
          * that the state machine can enter its correct state BEFORE we start processing
          * the events
          */
         CountDownLatch latch = new CountDownLatch(1);
-        
+
         new Thread(() -> {
             synchronized (curLocker) {
                 if (assignments == null) {
+                    latch.countDown();
                     return;
                 }
 
@@ -779,11 +804,11 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
                 while ((ev = eventq.poll()) != null) {
                     handle(ev);
                 }
-                
+
                 latch.countDown();
             }
         }).start();
-        
+
         return latch;
     }
 
@@ -876,12 +901,13 @@ public class PoolingManagerImpl implements PoolingManager, TopicListener {
         /**
          * Creates a DMaaP manager.
          * 
-         * @param props properties used to configure the manager
+         * @param topic name of the internal DMaaP topic
+         * @param props properties used to configure DMaaP
          * @return a new DMaaP manager
          * @throws PoolingFeatureException if an error occurs
          */
-        public DmaapManager makeDmaapManager(PoolingProperties props) throws PoolingFeatureException {
-            return new DmaapManager(props.getPoolingTopic(), props.getSource());
+        public DmaapManager makeDmaapManager(String topic, Properties props) throws PoolingFeatureException {
+            return new DmaapManager(topic, props);
         }
 
         /**
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java
index 4e8de0d0..54319423 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/PoolingProperties.java
@@ -51,12 +51,12 @@ public class PoolingProperties extends SpecPropertyConfiguration {
     public static final String FEATURE_ENABLED = PREFIX + "{?.}enabled";
     public static final String OFFLINE_LIMIT = PREFIX + "{?.}offline.queue.limit";
     public static final String OFFLINE_AGE_MS = PREFIX + "{?.}offline.queue.age.milliseconds";
+    public static final String OFFLINE_PUB_WAIT_MS = PREFIX + "{?.}offline.publish.wait.milliseconds";
     public static final String START_HEARTBEAT_MS = PREFIX + "{?.}start.heartbeat.milliseconds";
     public static final String REACTIVATE_MS = PREFIX + "{?.}reactivate.milliseconds";
     public static final String IDENTIFICATION_MS = PREFIX + "{?.}identification.milliseconds";
     public static final String ACTIVE_HEARTBEAT_MS = PREFIX + "{?.}active.heartbeat.milliseconds";
     public static final String INTER_HEARTBEAT_MS = PREFIX + "{?.}inter.heartbeat.milliseconds";
-    public static final String OFFLINE_PUB_WAIT_MS = PREFIX + "{?.}offline.publish.wait.milliseconds";
 
     /**
      * Type of item that the extractors will be extracting.
@@ -94,10 +94,17 @@ public class PoolingProperties extends SpecPropertyConfiguration {
     private long offlineAgeMs;
 
     /**
+     * Time, in milliseconds, to wait for an "Offline" message to be published
+     * to DMaaP.
+     */
+    @Property(name = OFFLINE_PUB_WAIT_MS, defaultValue = "3000")
+    private long offlinePubWaitMs;
+
+    /**
      * Time, in milliseconds, to wait for this host's heart beat during the
      * start-up state.
      */
-    @Property(name = START_HEARTBEAT_MS, defaultValue = "50000")
+    @Property(name = START_HEARTBEAT_MS, defaultValue = "100000")
     private long startHeartbeatMs;
 
     /**
@@ -123,19 +130,12 @@ public class PoolingProperties extends SpecPropertyConfiguration {
 
     /**
      * Time, in milliseconds, to wait between heart beat generations during
-     * the active state.
+     * the active and start-up states.
      */
     @Property(name = INTER_HEARTBEAT_MS, defaultValue = "15000")
     private long interHeartbeatMs;
 
     /**
-     * Time, in milliseconds, to wait for an "Offline" message to be published
-     * to DMaaP.
-     */
-    @Property(name = OFFLINE_PUB_WAIT_MS, defaultValue = "3000")
-    private long offlinePubWaitMs;
-
-    /**
      * @param controllerName the name of the controller
      * @param props set of properties used to configure this
      * @throws PropertyException if an error occurs
@@ -163,6 +163,10 @@ public class PoolingProperties extends SpecPropertyConfiguration {
         return offlineAgeMs;
     }
 
+    public long getOfflinePubWaitMs() {
+        return offlinePubWaitMs;
+    }
+
     public long getStartHeartbeatMs() {
         return startHeartbeatMs;
     }
@@ -182,8 +186,4 @@ public class PoolingProperties extends SpecPropertyConfiguration {
     public long getInterHeartbeatMs() {
         return interHeartbeatMs;
     }
-
-    public long getOfflinePubWaitMs() {
-        return offlinePubWaitMs;
-    }
 }
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/SpecProperties.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/SpecProperties.java
index 31b4e614..b62ea0a7 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/SpecProperties.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/SpecProperties.java
@@ -40,7 +40,8 @@ public class SpecProperties extends Properties {
 
     /**
      * 
-     * @param prefix the property name prefix that appears before any specialization
+     * @param prefix the property name prefix that appears before any specialization, may
+     *        be ""
      * @param specialization the property name specialization (e.g., session name)
      */
     public SpecProperties(String prefix, String specialization) {
@@ -50,7 +51,8 @@ public class SpecProperties extends Properties {
 
     /**
      * 
-     * @param prefix the property name prefix that appears before any specialization
+     * @param prefix the property name prefix that appears before any specialization, may
+     *        be ""
      * @param specialization the property name specialization (e.g., session name)
      * @param props the default properties
      */
@@ -68,7 +70,7 @@ public class SpecProperties extends Properties {
      * @return the text, with a trailing "."
      */
     private static String withTrailingDot(String text) {
-        return text.endsWith(".") ? text : text + ".";
+        return text.isEmpty() || text.endsWith(".") ? text : text + ".";
     }
 
     /**
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/extractor/ClassExtractors.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/extractor/ClassExtractors.java
index 2752ca8c..ccca77be 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/extractor/ClassExtractors.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/extractor/ClassExtractors.java
@@ -374,7 +374,7 @@ public class ClassExtractors {
 
             } catch (NoSuchMethodException expected) {
                 // no getXxx() method, maybe there's a field by this name
-                logger.debug("no method {} in {}", nm, clazz.getName());
+                logger.debug("no method {} in {}", nm, clazz.getName(), expected);
                 return null;
 
             } catch (SecurityException e) {
@@ -445,7 +445,7 @@ public class ClassExtractors {
 
             } catch (NoSuchFieldException expected) {
                 // no field by this name - try super class & interfaces
-                logger.debug("no field {} in {}", name, clazz.getName());
+                logger.debug("no field {} in {}", name, clazz.getName(), expected);
 
             } catch (SecurityException e) {
                 throw new ExtractorException("inaccessible field " + clazz + "." + name, e);
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java
index b0a36cd9..8f0a902a 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ActiveState.java
@@ -95,18 +95,19 @@ public class ActiveState extends ProcessingState {
         if ((succHost = assigned.higher(getHost())) == null) {
             // wrapped around - successor is the first host in the set
             succHost = assigned.first();
-            logger.info("this host's successor is {} on topic {}", succHost, getTopic());
         }
+        logger.info("this host's successor is {} on topic {}", succHost, getTopic());        
 
         if ((predHost = assigned.lower(getHost())) == null) {
             // wrapped around - predecessor is the last host in the set
             predHost = assigned.last();
-            logger.info("this host's predecessor is {} on topic {}", predHost, getTopic());
         }
+        logger.info("this host's predecessor is {} on topic {}", predHost, getTopic());
     }
 
     @Override
     public void start() {
+        super.start();
         addTimers();
         genHeartbeat();
     }
@@ -120,7 +121,7 @@ public class ActiveState extends ProcessingState {
         /*
          * heart beat generator
          */
-        long genMs = getProperties().getActiveHeartbeatMs();
+        long genMs = getProperties().getInterHeartbeatMs();
 
         scheduleWithFixedDelay(genMs, genMs, () -> {
             genHeartbeat();
@@ -130,9 +131,9 @@ public class ActiveState extends ProcessingState {
         /*
          * my heart beat checker
          */
-        long interMs = getProperties().getInterHeartbeatMs();
+        long waitMs = getProperties().getActiveHeartbeatMs();
 
-        scheduleWithFixedDelay(interMs, interMs, () -> {
+        scheduleWithFixedDelay(waitMs, waitMs, () -> {
             if (myHeartbeatSeen) {
                 myHeartbeatSeen = false;
                 return null;
@@ -141,7 +142,7 @@ public class ActiveState extends ProcessingState {
             // missed my heart beat
             logger.error("missed my heartbeat on topic {}", getTopic());
 
-            return internalTopicFailed();
+            return missedHeartbeat();
         });
 
         /*
@@ -149,7 +150,7 @@ public class ActiveState extends ProcessingState {
          */
         if (!predHost.isEmpty()) {
 
-            scheduleWithFixedDelay(interMs, interMs, () -> {
+            scheduleWithFixedDelay(waitMs, waitMs, () -> {
                 if (predHeartbeatSeen) {
                     predHeartbeatSeen = false;
                     return null;
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java
index 6be2fb84..da044252 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/InactiveState.java
@@ -44,24 +44,17 @@ public class InactiveState extends State {
 
     @Override
     public void start() {
-
         super.start();
-
         schedule(getProperties().getReactivateMs(), () -> goStart());
     }
 
     @Override
     public State process(Leader msg) {
-        if(isValid(msg)) {
+        if (isValid(msg)) {
             logger.info("received Leader message from {} on topic {}", msg.getSource(), getTopic());
-            startDistributing(msg.getAssignments());
-            
-            if(msg.getAssignments().hasAssignment(getHost())) {
-                logger.info("received Leader message on topic {}", getTopic());
-                return goActive();
-            }
+            return goActive(msg.getAssignments());
         }
-        
+
         return null;
     }
 
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java
index 1e9bb581..e9dc0324 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/ProcessingState.java
@@ -74,24 +74,6 @@ public class ProcessingState extends State {
     }
 
     /**
-     * Goes active with a new set of assignments.
-     * 
-     * @param asgn new assignments
-     * @return the new state, either Active or Inactive, depending on whether or not this
-     *         host has an assignment
-     */
-    protected State goActive(BucketAssignments asgn) {
-        startDistributing(asgn);
-
-        if (asgn.hasAssignment(getHost())) {
-            return goActive();
-
-        } else {
-            return goInactive();
-        }
-    }
-
-    /**
      * Generates an Identification message and goes to the query state.
      */
     @Override
@@ -154,11 +136,11 @@ public class ProcessingState extends State {
         }
 
         Leader msg = makeLeader(alive);
-        publish(msg);
+        logger.info("{}/{} hosts have an assignment", msg.getAssignments().getAllHosts().size(), alive.size());
 
-        setAssignments(msg.getAssignments());
+        publish(msg);
 
-        return goActive();
+        return goActive(msg.getAssignments());
     }
 
     /**
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java
index 9045165b..1a4da150 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/QueryState.java
@@ -63,7 +63,6 @@ public class QueryState extends ProcessingState {
 
     @Override
     public void start() {
-
         super.start();
 
         // start identification timer
@@ -85,39 +84,21 @@ public class QueryState extends ProcessingState {
             if (!sawSelfIdent) {
                 // didn't see our identification
                 logger.error("missed our own Ident message on topic {}", getTopic());
-                return internalTopicFailed();
+                return missedHeartbeat();
 
             } else if (isLeader()) {
                 // "this" host is the new leader
                 logger.info("this host is the new leader for topic {}", getTopic());
                 return becomeLeader(alive);
 
-            } else if (hasAssignment()) {
-                /*
-                 * this host is not the new leader, but it does have an assignment -
-                 * return to the active state while we wait for the leader
-                 */
-                logger.info("no new leader on topic {}", getTopic());
-                return goActive();
-
             } else {
-                // not the leader and no assignment yet
+                // not the leader - return to previous state
                 logger.info("no new leader on topic {}", getTopic());
-                return goInactive();
+                return goActive(getAssignments());
             }
         });
     }
 
-    /**
-     * Determines if this host has an assignment in the CURRENT assignments.
-     * 
-     * @return {@code true} if this host has an assignment, {@code false} otherwise
-     */
-    protected boolean hasAssignment() {
-        BucketAssignments asgn = getAssignments();
-        return (asgn != null && asgn.hasAssignment(getHost()));
-    }
-
     @Override
     public State goQuery() {
         return null;
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/StartState.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/StartState.java
index a978e245..3068cfc9 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/StartState.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/StartState.java
@@ -67,8 +67,22 @@ public class StartState extends State {
 
         super.start();
 
-        publish(getHost(), makeHeartbeat(hbTimestampMs));
+        Heartbeat hb = makeHeartbeat(hbTimestampMs);
+        publish(getHost(), hb);
 
+        /*
+         * heart beat generator
+         */
+        long genMs = getProperties().getInterHeartbeatMs();
+
+        scheduleWithFixedDelay(genMs, genMs, () -> {
+            publish(getHost(), hb);
+            return null;
+        });
+
+        /*
+         * my heart beat checker
+         */
         schedule(getProperties().getStartHeartbeatMs(), () -> {
             logger.error("missed heartbeat on topic {}", getTopic());
             return internalTopicFailed();
diff --git a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/State.java b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/State.java
index 421b9225..54e93230 100644
--- a/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/State.java
+++ b/feature-pooling-dmaap/src/main/java/org/onap/policy/drools/pooling/state/State.java
@@ -113,12 +113,21 @@ public abstract class State {
     }
 
     /**
-     * Transitions to the "active" state.
+     * Goes active with a new set of assignments.
      * 
-     * @return the new state
+     * @param asgn new assignments
+     * @return the new state, either Active or Inactive, depending on whether or not this
+     *         host has an assignment
      */
-    public final State goActive() {
-        return mgr.goActive();
+    protected State goActive(BucketAssignments asgn) {
+        startDistributing(asgn);
+
+        if (asgn.hasAssignment(getHost())) {
+            return mgr.goActive();
+
+        } else {
+            return goInactive();
+        }
     }
 
     /**
@@ -322,7 +331,21 @@ public abstract class State {
     }
 
     /**
-     * Indicates that the internal topic failed.
+     * Indicates that we failed to see our own heartbeat; must be a problem with the
+     * internal topic.
+     * 
+     * @return a new {@link StartState}
+     */
+    protected final State missedHeartbeat() {
+        publish(makeOffline());
+        mgr.startDistributing(null);
+
+        return mgr.goStart();
+    }
+
+    /**
+     * Indicates that the internal topic failed; this should only be invoked from the
+     * StartState.
      * 
      * @return a new {@link InactiveState}
      */