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>

25 files changed, 1201 insertions, 391 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}
      */
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/DmaapManagerTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/DmaapManagerTest.java
index a91671fd..29dc15e4 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/DmaapManagerTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/DmaapManagerTest.java
@@ -211,8 +211,8 @@ public class DmaapManagerTest {
         // force exception when it starts
         doThrow(new IllegalStateException("expected")).when(sink).start();
 
-        expectException("startPublisher,start", xxx -> mgr.startPublisher());
-        expectException("startPublisher,publish", xxx -> mgr.publish(MSG));
+        expectException("startPublisher,start", () -> mgr.startPublisher());
+        expectException("startPublisher,publish", () -> mgr.publish(MSG));
 
         // allow it to succeed this time
         reset(sink);
@@ -336,7 +336,7 @@ public class DmaapManagerTest {
     @Test
     public void testPublish() throws PoolingFeatureException {
         // cannot publish before starting
-        expectException("publish,pre", xxx -> mgr.publish(MSG));
+        expectException("publish,pre", () -> mgr.publish(MSG));
 
         mgr.startPublisher();
 
@@ -352,7 +352,7 @@ public class DmaapManagerTest {
 
         // stop and verify we can no longer publish
         mgr.stopPublisher(0);
-        expectException("publish,stopped", xxx -> mgr.publish(MSG));
+        expectException("publish,stopped", () -> mgr.publish(MSG));
     }
 
     @Test(expected = PoolingFeatureException.class)
@@ -377,7 +377,7 @@ public class DmaapManagerTest {
 
     private void expectException(String testnm, VFunction func) {
         try {
-            func.apply(null);
+            func.apply();
             fail(testnm + " missing exception");
 
         } catch (PoolingFeatureException expected) {
@@ -387,6 +387,6 @@ public class DmaapManagerTest {
 
     @FunctionalInterface
     public static interface VFunction {
-        public void apply(Void arg) throws PoolingFeatureException;
+        public void apply() throws PoolingFeatureException;
     }
 }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest.java
index 13d70f52..cc588384 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest.java
@@ -26,13 +26,13 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.onap.policy.common.utils.properties.SpecPropertyConfiguration.specialize;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Deque;
 import java.util.IdentityHashMap;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.TreeMap;
 import java.util.concurrent.BlockingQueue;
@@ -43,13 +43,11 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
@@ -71,7 +69,15 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 
 /**
  * End-to-end tests of the pooling feature. Launches one or more "hosts", each one having
- * its own feature object.
+ * its own feature object. Uses real feature objects. However, the following are not:
+ * <dl>
+ * <dt>DMaaP sources and sinks</dt>
+ * <dd>simulated using queues. There is one queue for the external topic, and one queue
+ * for each host's internal topic. Messages published to the "admin" channel are simply
+ * sent to all of the hosts' internal topic queues</dd>
+ * <dt>PolicyEngine, PolicyController, DroolsController</dt>
+ * <dd>mocked</dd>
+ * </dl>
  */
 public class FeatureTest {
 
@@ -92,26 +98,15 @@ public class FeatureTest {
      */
     private static final String CONTROLLER1 = "controller.one";
 
-    // private static final long STD_HEARTBEAT_WAIT_MS = 100;
-    // private static final long STD_REACTIVATE_WAIT_MS = 200;
-    // private static final long STD_IDENTIFICATION_MS = 60;
-    // private static final long STD_ACTIVE_HEARTBEAT_MS = 5;
-    // private static final long STD_INTER_HEARTBEAT_MS = 50;
-    // private static final long STD_OFFLINE_PUB_WAIT_MS = 2;
-    // private static final long POLL_MS = 2;
-    // private static final long INTER_POLL_MS = 2;
-    // private static final long EVENT_WAIT_SEC = 5;
-
-    // use these to slow things down
-    private static final long STD_HEARTBEAT_WAIT_MS = 5000;
-    private static final long STD_REACTIVATE_WAIT_MS = 10000;
-    private static final long STD_IDENTIFICATION_MS = 10000;
-    private static final long STD_ACTIVE_HEARTBEAT_MS = 5000;
-    private static final long STD_INTER_HEARTBEAT_MS = 12000;
-    private static final long STD_OFFLINE_PUB_WAIT_MS = 2;
-    private static final long POLL_MS = 2;
-    private static final long INTER_POLL_MS = 2000;
-    private static final long EVENT_WAIT_SEC = 1000;
+    private static long stdReactivateWaitMs = 200;
+    private static long stdIdentificationMs = 60;
+    private static long stdStartHeartbeatMs = 60;
+    private static long stdActiveHeartbeatMs = 50;
+    private static long stdInterHeartbeatMs = 5;
+    private static long stdOfflinePubWaitMs = 2;
+    private static long stdPollMs = 2;
+    private static long stdInterPollMs = 2;
+    private static long stdEventWaitSec = 10;
 
     // these are saved and restored on exit from this test class
     private static PoolingFeature.Factory saveFeatureFactory;
@@ -128,6 +123,8 @@ public class FeatureTest {
         saveFeatureFactory = PoolingFeature.getFactory();
         saveManagerFactory = PoolingManagerImpl.getFactory();
         saveDmaapFactory = DmaapManager.getFactory();
+        
+        // note: invoke runSlow() to slow things down
     }
 
     @AfterClass
@@ -149,51 +146,35 @@ public class FeatureTest {
         }
     }
 
-    @Ignore
     @Test
     public void test_SingleHost() throws Exception {
-        int nmessages = 70;
-
-        ctx = new Context(nmessages);
-
-        ctx.addHost();
-        ctx.startHosts();
-
-        for (int x = 0; x < nmessages; ++x) {
-            ctx.offerExternal(makeMessage(x));
-        }
-
-        ctx.awaitEvents(EVENT_WAIT_SEC, TimeUnit.SECONDS);
-
-        assertEquals(0, ctx.getDecodeErrors());
-        assertEquals(0, ctx.getRemainingEvents());
-        ctx.checkAllSawAMsg();
+        run(70, 1);
     }
 
-    @Ignore
     @Test
     public void test_TwoHosts() throws Exception {
-        int nmessages = 200;
+        run(200, 2);
+    }
+
+    @Test
+    public void test_ThreeHosts() throws Exception {
+        run(200, 3);
+    }
 
+    private void run(int nmessages, int nhosts) throws Exception {
         ctx = new Context(nmessages);
 
-        ctx.addHost();
-        ctx.addHost();
+        for (int x = 0; x < nhosts; ++x) {
+            ctx.addHost();
+        }
+
         ctx.startHosts();
 
         for (int x = 0; x < nmessages; ++x) {
             ctx.offerExternal(makeMessage(x));
         }
 
-        // wait for all hosts to have time to process a few messages
-        Thread.sleep(STD_ACTIVE_HEARTBEAT_MS + INTER_POLL_MS * 3);
-
-        // pause a topic for a bit
-//        ctx.pauseTopic();
-
-        // now we'll see if it recovers
-
-        ctx.awaitEvents(EVENT_WAIT_SEC, TimeUnit.SECONDS);
+        ctx.awaitEvents(stdEventWaitSec, TimeUnit.SECONDS);
 
         assertEquals(0, ctx.getDecodeErrors());
         assertEquals(0, ctx.getRemainingEvents());
@@ -203,6 +184,21 @@ public class FeatureTest {
     private String makeMessage(int reqnum) {
         return "{\"reqid\":\"req" + reqnum + "\", \"data\":\"hello " + reqnum + "\"}";
     }
+    
+    /**
+     * Invoke this to slow the timers down.
+     */
+    protected static void runSlow() {
+         stdReactivateWaitMs = 10000;
+         stdIdentificationMs = 10000;
+         stdStartHeartbeatMs = 15000;
+         stdActiveHeartbeatMs = 12000;
+         stdInterHeartbeatMs = 5000;
+         stdOfflinePubWaitMs = 2;
+         stdPollMs = 2;
+         stdInterPollMs = 2000;
+         stdEventWaitSec = 1000;
+    }
 
     /**
      * Context used for a single test case.
@@ -244,12 +240,6 @@ public class FeatureTest {
         private final CountDownLatch eventCounter;
 
         /**
-         * Maps host name to its topic source. This must be in sorted order so we can
-         * identify the source for the host with the higher name.
-         */
-        private TreeMap<String, TopicSourceImpl> host2topic = new TreeMap<>();
-
-        /**
          * The current host. Set by {@link #withHost(Host, VoidFunction)} and used by
          * {@link #getCurrentHost()}.
          */
@@ -280,9 +270,14 @@ public class FeatureTest {
 
         /**
          * Creates and adds a new host to the context.
+         * 
+         * @return the new Host
          */
-        public void addHost() {
-            hosts.add(new Host(this));
+        public Host addHost() {
+            Host host = new Host(this);
+            hosts.add(host);
+
+            return host;
         }
 
         /**
@@ -443,26 +438,6 @@ public class FeatureTest {
         }
 
         /**
-         * Associates a host with a topic.
-         * 
-         * @param host
-         * @param topic
-         */
-        public void addTopicSource(String host, TopicSourceImpl topic) {
-            host2topic.put(host, topic);
-        }
-
-        /**
-         * Pauses the last topic source long enough to miss a heart beat.
-         */
-        public void pauseTopic() {
-            Entry<String, TopicSourceImpl> ent = host2topic.lastEntry();
-            if (ent != null) {
-                ent.getValue().pause(STD_ACTIVE_HEARTBEAT_MS);
-            }
-        }
-
-        /**
          * Gets the current host, provided this is used from within a call to
          * {@link #withHost(Host, VoidFunction)}.
          * 
@@ -527,12 +502,10 @@ public class FeatureTest {
         }
 
         /**
-         * Gets the host name. This should only be invoked within {@link #start()}.
-         * 
          * @return the host name
          */
         public String getName() {
-            return PoolingManagerImpl.getLastHost();
+            return feature.getHost();
         }
 
         /**
@@ -759,11 +732,6 @@ public class FeatureTest {
         private AtomicReference<Pair<CountDownLatch, CountDownLatch>> pair = new AtomicReference<>(null);
 
         /**
-         * Time, in milliseconds, to pause before polling for more messages.
-         */
-        private AtomicLong pauseTimeMs = new AtomicLong(0);
-
-        /**
          * 
          * @param context
          * @param internal {@code true} if to read from the internal topic, {@code false}
@@ -771,12 +739,8 @@ public class FeatureTest {
          */
         public TopicSourceImpl(Context context, boolean internal) {
             if (internal) {
-                Host host = context.getCurrentHost();
-
                 this.topic = INTERNAL_TOPIC;
-                this.queue = host.getInternalQueue();
-
-                context.addTopicSource(host.getName(), this);
+                this.queue = context.getCurrentHost().getInternalQueue();
 
             } else {
                 this.topic = EXTERNAL_TOPIC;
@@ -809,11 +773,12 @@ public class FeatureTest {
 
             reregister(newPair);
 
-            new Thread(() -> {
+            Thread thread = new Thread(() -> {
+
                 try {
                     do {
                         processMessages(newPair.first(), listener);
-                    } while (!newPair.first().await(INTER_POLL_MS, TimeUnit.MILLISECONDS));
+                    } while (!newPair.first().await(stdInterPollMs, TimeUnit.MILLISECONDS));
 
                     logger.info("topic source thread completed");
 
@@ -827,7 +792,10 @@ public class FeatureTest {
 
                 newPair.second().countDown();
 
-            }).start();
+            });
+
+            thread.setDaemon(true);
+            thread.start();
         }
 
         /**
@@ -879,19 +847,7 @@ public class FeatureTest {
         }
 
         /**
-         * Indicates that {@link #processMessages(CountDownLatch, TopicListener)} should
-         * pause a bit.
-         * 
-         * @param timeMs time, in milliseconds, to pause
-         */
-        public void pause(long timeMs) {
-            pauseTimeMs.set(timeMs);
-        }
-
-        /**
-         * Polls for messages from the topic and offers them to the listener. If
-         * {@link #pauseTimeMs} is non-zero, then it pauses for the specified time and
-         * then immediately returns.
+         * Polls for messages from the topic and offers them to the listener.
          * 
          * @param stopped triggered if processing should stop
          * @param listener
@@ -901,14 +857,7 @@ public class FeatureTest {
 
             for (int x = 0; x < 5 && stopped.getCount() > 0; ++x) {
 
-                long ptm = pauseTimeMs.getAndSet(0);
-                if (ptm != 0) {
-                    logger.warn("pause processing");
-                    stopped.await(ptm, TimeUnit.MILLISECONDS);
-                    return;
-                }
-
-                String msg = queue.poll(POLL_MS, TimeUnit.MILLISECONDS);
+                String msg = queue.poll(stdPollMs, TimeUnit.MILLISECONDS);
                 if (msg == null) {
                     return;
                 }
@@ -1038,18 +987,20 @@ public class FeatureTest {
 
             props.setProperty(PoolingProperties.PROP_EXTRACTOR_PREFIX + ".java.util.Map", "${reqid}");
 
-            props.setProperty("pooling." + CONTROLLER1 + ".topic", INTERNAL_TOPIC);
-            props.setProperty("pooling." + CONTROLLER1 + ".enabled", "true");
-            props.setProperty("pooling." + CONTROLLER1 + ".offline.queue.limit", "10000");
-            props.setProperty("pooling." + CONTROLLER1 + ".offline.queue.age.milliseconds", "1000000");
-            props.setProperty("pooling." + CONTROLLER1 + ".start.heartbeat.milliseconds", "" + STD_HEARTBEAT_WAIT_MS);
-            props.setProperty("pooling." + CONTROLLER1 + ".reactivate.milliseconds", "" + STD_REACTIVATE_WAIT_MS);
-            props.setProperty("pooling." + CONTROLLER1 + ".identification.milliseconds", "" + STD_IDENTIFICATION_MS);
-            props.setProperty("pooling." + CONTROLLER1 + ".active.heartbeat.milliseconds",
-                            "" + STD_ACTIVE_HEARTBEAT_MS);
-            props.setProperty("pooling." + CONTROLLER1 + ".inter.heartbeat.milliseconds", "" + STD_INTER_HEARTBEAT_MS);
-            props.setProperty("pooling." + CONTROLLER1 + ".offline.publish.wait.milliseconds",
-                            "" + STD_OFFLINE_PUB_WAIT_MS);
+            props.setProperty(specialize(PoolingProperties.FEATURE_ENABLED, CONTROLLER1), "true");
+            props.setProperty(specialize(PoolingProperties.POOLING_TOPIC, CONTROLLER1), INTERNAL_TOPIC);
+            props.setProperty(specialize(PoolingProperties.OFFLINE_LIMIT, CONTROLLER1), "10000");
+            props.setProperty(specialize(PoolingProperties.OFFLINE_AGE_MS, CONTROLLER1), "1000000");
+            props.setProperty(specialize(PoolingProperties.OFFLINE_PUB_WAIT_MS, CONTROLLER1),
+                            "" + stdOfflinePubWaitMs);
+            props.setProperty(specialize(PoolingProperties.START_HEARTBEAT_MS, CONTROLLER1),
+                            "" + stdStartHeartbeatMs);
+            props.setProperty(specialize(PoolingProperties.REACTIVATE_MS, CONTROLLER1), "" + stdReactivateWaitMs);
+            props.setProperty(specialize(PoolingProperties.IDENTIFICATION_MS, CONTROLLER1), "" + stdIdentificationMs);
+            props.setProperty(specialize(PoolingProperties.ACTIVE_HEARTBEAT_MS, CONTROLLER1),
+                            "" + stdActiveHeartbeatMs);
+            props.setProperty(specialize(PoolingProperties.INTER_HEARTBEAT_MS, CONTROLLER1),
+                            "" + stdInterHeartbeatMs);
 
             return props;
         }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java
new file mode 100644
index 00000000..84449e75
--- /dev/null
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java
@@ -0,0 +1,708 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP
+ * ================================================================================
+ * Copyright (C) 2018 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.drools.pooling;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.onap.policy.common.utils.properties.SpecPropertyConfiguration.specialize;
+import java.io.IOException;
+import java.util.Deque;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.Properties;
+import java.util.TreeMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.onap.policy.drools.controller.DroolsController;
+import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.drools.event.comm.TopicEndpoint;
+import org.onap.policy.drools.event.comm.TopicListener;
+import org.onap.policy.drools.event.comm.TopicSink;
+import org.onap.policy.drools.event.comm.TopicSource;
+import org.onap.policy.drools.properties.PolicyProperties;
+import org.onap.policy.drools.system.PolicyController;
+import org.onap.policy.drools.system.PolicyEngine;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * End-to-end tests of the pooling feature. Launches one or more "hosts", each one having
+ * its own feature object. Uses real feature objects, as well as real DMaaP sources and
+ * sinks. However, the following are not:
+ * <dl>
+ * <dt>PolicyEngine, PolicyController, DroolsController</dt>
+ * <dd>mocked</dd>
+ * </dl>
+ * 
+ * <p>
+ * The following fields must be set before executing this:
+ * <ul>
+ * <li>UEB_SERVERS</li>
+ * <li>INTERNAL_TOPIC</li>
+ * <li>EXTERNAL_TOPIC</li>
+ * </ul>
+ */
+public class FeatureTest2 {
+
+    private static final Logger logger = LoggerFactory.getLogger(FeatureTest2.class);
+
+    /**
+     * UEB servers for both internal & external topics.
+     */
+    private static final String UEB_SERVERS = "";
+
+    /**
+     * Name of the topic used for inter-host communication.
+     */
+    private static final String INTERNAL_TOPIC = "";
+
+    /**
+     * Name of the topic from which "external" events "arrive".
+     */
+    private static final String EXTERNAL_TOPIC = "";
+
+    /**
+     * Consumer group to use when polling the external topic.
+     */
+    private static final String EXTERNAL_GROUP = FeatureTest2.class.getName();
+
+    /**
+     * Name of the controller.
+     */
+    private static final String CONTROLLER1 = "controller.one";
+
+    /**
+     * Maximum number of items to fetch from DMaaP in a single poll.
+     */
+    private static final String FETCH_LIMIT = "5";
+
+    private static final long STD_REACTIVATE_WAIT_MS = 10000;
+    private static final long STD_IDENTIFICATION_MS = 10000;
+    private static final long STD_START_HEARTBEAT_MS = 15000;
+    private static final long STD_ACTIVE_HEARTBEAT_MS = 12000;
+    private static final long STD_INTER_HEARTBEAT_MS = 5000;
+    private static final long STD_OFFLINE_PUB_WAIT_MS = 2;
+    private static final long EVENT_WAIT_SEC = 15;
+
+    // these are saved and restored on exit from this test class
+    private static PoolingFeature.Factory saveFeatureFactory;
+    private static PoolingManagerImpl.Factory saveManagerFactory;
+
+    /**
+     * Sink for external DMaaP topic.
+     */
+    private static TopicSink externalSink;
+
+    /**
+     * Context for the current test case.
+     */
+    private Context ctx;
+
+
+    @BeforeClass
+    public static void setUpBeforeClass() {
+        saveFeatureFactory = PoolingFeature.getFactory();
+        saveManagerFactory = PoolingManagerImpl.getFactory();
+
+        Properties props = makeSinkProperties(EXTERNAL_TOPIC);
+        externalSink = TopicEndpoint.manager.addTopicSinks(props).get(0);
+        externalSink.start();
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() {
+        PoolingFeature.setFactory(saveFeatureFactory);
+        PoolingManagerImpl.setFactory(saveManagerFactory);
+
+        externalSink.stop();
+    }
+
+    @Before
+    public void setUp() {
+        ctx = null;
+    }
+
+    @After
+    public void tearDown() {
+        if (ctx != null) {
+            ctx.destroy();
+        }
+    }
+
+    @Ignore
+    @Test
+    public void test_SingleHost() throws Exception {
+        run(70, 1);
+    }
+
+    @Ignore
+    @Test
+    public void test_TwoHosts() throws Exception {
+        run(200, 2);
+    }
+
+    @Ignore
+    @Test
+    public void test_ThreeHosts() throws Exception {
+        run(200, 3);
+    }
+
+    private void run(int nmessages, int nhosts) throws Exception {
+        ctx = new Context(nmessages);
+
+        for (int x = 0; x < nhosts; ++x) {
+            ctx.addHost();
+        }
+
+        ctx.startHosts();
+
+        ctx.awaitEvents(STD_IDENTIFICATION_MS * 2, TimeUnit.MILLISECONDS);
+
+        for (int x = 0; x < nmessages; ++x) {
+            ctx.offerExternal(makeMessage(x));
+        }
+
+        ctx.awaitEvents(EVENT_WAIT_SEC, TimeUnit.SECONDS);
+
+        assertEquals(0, ctx.getDecodeErrors());
+        assertEquals(0, ctx.getRemainingEvents());
+        ctx.checkAllSawAMsg();
+    }
+
+    private String makeMessage(int reqnum) {
+        return "{\"reqid\":\"req" + reqnum + "\", \"data\":\"hello " + reqnum + "\"}";
+    }
+
+    private static Properties makeSinkProperties(String topic) {
+        Properties props = new Properties();
+
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SINK_TOPICS, topic);
+
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_TOPIC_SERVERS_SUFFIX, UEB_SERVERS);
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX, "0");
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_MANAGED_SUFFIX, "false");
+
+        return props;
+    }
+
+    private static Properties makeSourceProperties(String topic) {
+        Properties props = new Properties();
+
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS, topic);
+
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_TOPIC_SERVERS_SUFFIX, UEB_SERVERS);
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX, FETCH_LIMIT);
+        props.setProperty(PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
+                        + PolicyProperties.PROPERTY_MANAGED_SUFFIX, "false");
+
+        if (EXTERNAL_TOPIC.equals(topic)) {
+            // consumer group is a constant
+            props.setProperty(
+                            PolicyProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
+                                            + PolicyProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX,
+                            EXTERNAL_GROUP);
+
+            // consumer instance is generated by the BusConsumer code
+        }
+
+        // else internal topic: feature populates info for internal topic
+
+        return props;
+    }
+
+    /**
+     * Context used for a single test case.
+     */
+    private static class Context {
+
+        private final FeatureFactory featureFactory;
+        private final ManagerFactory managerFactory;
+
+        /**
+         * Hosts that have been added to this context.
+         */
+        private final Deque<Host> hosts = new LinkedList<>();
+
+        /**
+         * Maps a drools controller to its policy controller.
+         */
+        private final IdentityHashMap<DroolsController, PolicyController> drools2policy = new IdentityHashMap<>();
+
+        /**
+         * Counts the number of decode errors.
+         */
+        private final AtomicInteger nDecodeErrors = new AtomicInteger(0);
+
+        /**
+         * Number of events we're still waiting to receive.
+         */
+        private final CountDownLatch eventCounter;
+
+        /**
+         * 
+         * @param nEvents number of events to be processed
+         */
+        public Context(int nEvents) {
+            featureFactory = new FeatureFactory(this);
+            managerFactory = new ManagerFactory(this);
+            eventCounter = new CountDownLatch(nEvents);
+
+            PoolingFeature.setFactory(featureFactory);
+            PoolingManagerImpl.setFactory(managerFactory);
+        }
+
+        /**
+         * Destroys the context, stopping any hosts that remain.
+         */
+        public void destroy() {
+            stopHosts();
+            hosts.clear();
+        }
+
+        /**
+         * Creates and adds a new host to the context.
+         * 
+         * @return the new Host
+         */
+        public Host addHost() {
+            Host host = new Host(this);
+            hosts.add(host);
+
+            return host;
+        }
+
+        /**
+         * Starts the hosts.
+         */
+        public void startHosts() {
+            hosts.forEach(host -> host.start());
+        }
+
+        /**
+         * Stops the hosts.
+         */
+        public void stopHosts() {
+            hosts.forEach(host -> host.stop());
+        }
+
+        /**
+         * Verifies that all hosts processed at least one message.
+         */
+        public void checkAllSawAMsg() {
+            int x = 0;
+            for (Host host : hosts) {
+                assertTrue("x=" + x, host.messageSeen());
+                ++x;
+            }
+        }
+
+        /**
+         * Offers an event to the external topic.
+         * 
+         * @param event
+         */
+        public void offerExternal(String event) {
+            externalSink.send(event);
+        }
+
+        /**
+         * Decodes an event.
+         * 
+         * @param event
+         * @return the decoded event, or {@code null} if it cannot be decoded
+         */
+        public Object decodeEvent(String event) {
+            return managerFactory.decodeEvent(null, null, event);
+        }
+
+        /**
+         * Associates a controller with its drools controller.
+         * 
+         * @param controller
+         * @param droolsController
+         */
+        public void addController(PolicyController controller, DroolsController droolsController) {
+            drools2policy.put(droolsController, controller);
+        }
+
+        /**
+         * @param droolsController
+         * @return the controller associated with a drools controller, or {@code null} if
+         *         it has no associated controller
+         */
+        public PolicyController getController(DroolsController droolsController) {
+            return drools2policy.get(droolsController);
+        }
+
+        /**
+         * 
+         * @return the number of decode errors so far
+         */
+        public int getDecodeErrors() {
+            return nDecodeErrors.get();
+        }
+
+        /**
+         * Increments the count of decode errors.
+         */
+        public void bumpDecodeErrors() {
+            nDecodeErrors.incrementAndGet();
+        }
+
+        /**
+         * 
+         * @return the number of events that haven't been processed
+         */
+        public long getRemainingEvents() {
+            return eventCounter.getCount();
+        }
+
+        /**
+         * Adds an event to the counter.
+         */
+        public void addEvent() {
+            eventCounter.countDown();
+        }
+
+        /**
+         * Waits, for a period of time, for all events to be processed.
+         * 
+         * @param time
+         * @param units
+         * @return {@code true} if all events have been processed, {@code false} otherwise
+         * @throws InterruptedException
+         */
+        public boolean awaitEvents(long time, TimeUnit units) throws InterruptedException {
+            return eventCounter.await(time, units);
+        }
+    }
+
+    /**
+     * Simulates a single "host".
+     */
+    private static class Host {
+
+        private final PoolingFeature feature = new PoolingFeature();
+
+        /**
+         * {@code True} if this host has processed a message, {@code false} otherwise.
+         */
+        private final AtomicBoolean sawMsg = new AtomicBoolean(false);
+
+        private final TopicSource externalSource;
+
+        // mock objects
+        private final PolicyEngine engine = mock(PolicyEngine.class);
+        private final ListenerController controller = mock(ListenerController.class);
+        private final DroolsController drools = mock(DroolsController.class);
+
+        /**
+         * 
+         * @param context
+         */
+        public Host(Context context) {
+
+            when(controller.getName()).thenReturn(CONTROLLER1);
+            when(controller.getDrools()).thenReturn(drools);
+
+            Properties props = makeSourceProperties(EXTERNAL_TOPIC);
+            externalSource = TopicEndpoint.manager.addTopicSources(props).get(0);
+
+            // stop consuming events if the controller stops
+            when(controller.stop()).thenAnswer(args -> {
+                externalSource.unregister(controller);
+                return true;
+            });
+
+            doAnswer(new MyExternalTopicListener(context, this)).when(controller).onTopicEvent(any(), any(), any());
+
+            context.addController(controller, drools);
+        }
+
+        /**
+         * Starts threads for the host so that it begins consuming from both the external
+         * "DMaaP" topic and its own internal "DMaaP" topic.
+         */
+        public void start() {
+            feature.beforeStart(engine);
+            feature.afterCreate(controller);
+
+            feature.beforeStart(controller);
+
+            // start consuming events from the external topic
+            externalSource.register(controller);
+
+            feature.afterStart(controller);
+        }
+
+        /**
+         * Stops the host's threads.
+         */
+        public void stop() {
+            feature.beforeStop(controller);
+            externalSource.unregister(controller);
+            feature.afterStop(controller);
+        }
+
+        /**
+         * Offers an event to the feature, before the policy controller handles it.
+         * 
+         * @param protocol
+         * @param topic2
+         * @param event
+         * @return {@code true} if the event was handled, {@code false} otherwise
+         */
+        public boolean beforeOffer(CommInfrastructure protocol, String topic2, String event) {
+            return feature.beforeOffer(controller, protocol, topic2, event);
+        }
+
+        /**
+         * Offers an event to the feature, after the policy controller handles it.
+         * 
+         * @param protocol
+         * @param topic
+         * @param event
+         * @param success
+         * @return {@code true} if the event was handled, {@code false} otherwise
+         */
+        public boolean afterOffer(CommInfrastructure protocol, String topic, String event, boolean success) {
+
+            return feature.afterOffer(controller, protocol, topic, event, success);
+        }
+
+        /**
+         * Offers an event to the feature, before the drools controller handles it.
+         * 
+         * @param fact
+         * @return {@code true} if the event was handled, {@code false} otherwise
+         */
+        public boolean beforeInsert(Object fact) {
+            return feature.beforeInsert(drools, fact);
+        }
+
+        /**
+         * Offers an event to the feature, after the drools controller handles it.
+         * 
+         * @param fact
+         * @param successInsert {@code true} if it was successfully inserted by the drools
+         *        controller, {@code false} otherwise
+         * @return {@code true} if the event was handled, {@code false} otherwise
+         */
+        public boolean afterInsert(Object fact, boolean successInsert) {
+            return feature.afterInsert(drools, fact, successInsert);
+        }
+
+        /**
+         * Indicates that a message was seen for this host.
+         */
+        public void sawMessage() {
+            sawMsg.set(true);
+        }
+
+        /**
+         * 
+         * @return {@code true} if a message was seen for this host, {@code false}
+         *         otherwise
+         */
+        public boolean messageSeen() {
+            return sawMsg.get();
+        }
+    }
+
+    /**
+     * Listener for the external topic. Simulates the actions taken by
+     * <i>AggregatedPolicyController.onTopicEvent</i>.
+     */
+    private static class MyExternalTopicListener implements Answer<Void> {
+
+        private final Context context;
+        private final Host host;
+
+        public MyExternalTopicListener(Context context, Host host) {
+            this.context = context;
+            this.host = host;
+        }
+
+        @Override
+        public Void answer(InvocationOnMock args) throws Throwable {
+            int i = 0;
+            CommInfrastructure commType = args.getArgument(i++);
+            String topic = args.getArgument(i++);
+            String event = args.getArgument(i++);
+
+            if (host.beforeOffer(commType, topic, event)) {
+                return null;
+            }
+
+            boolean result;
+            Object fact = context.decodeEvent(event);
+
+            if (fact == null) {
+                result = false;
+                context.bumpDecodeErrors();
+
+            } else {
+                result = true;
+
+                if (!host.beforeInsert(fact)) {
+                    // feature did not handle it so we handle it here
+                    host.afterInsert(fact, result);
+
+                    host.sawMessage();
+                    context.addEvent();
+                }
+            }
+
+            host.afterOffer(commType, topic, event, result);
+            return null;
+        }
+    }
+
+    /**
+     * Simulator for the feature-level factory.
+     */
+    private static class FeatureFactory extends PoolingFeature.Factory {
+
+        private final Context context;
+
+        /**
+         * 
+         * @param context
+         */
+        public FeatureFactory(Context context) {
+            this.context = context;
+
+            /*
+             * Note: do NOT extract anything from "context" at this point, because it
+             * hasn't been fully initialized yet
+             */
+        }
+
+        @Override
+        public Properties getProperties(String featName) {
+            Properties props = new Properties();
+
+            props.setProperty(PoolingProperties.PROP_EXTRACTOR_PREFIX + ".java.util.Map", "${reqid}");
+
+            props.setProperty(specialize(PoolingProperties.FEATURE_ENABLED, CONTROLLER1), "true");
+            props.setProperty(specialize(PoolingProperties.POOLING_TOPIC, CONTROLLER1), INTERNAL_TOPIC);
+            props.setProperty(specialize(PoolingProperties.OFFLINE_LIMIT, CONTROLLER1), "10000");
+            props.setProperty(specialize(PoolingProperties.OFFLINE_AGE_MS, CONTROLLER1), "1000000");
+            props.setProperty(specialize(PoolingProperties.OFFLINE_PUB_WAIT_MS, CONTROLLER1),
+                            "" + STD_OFFLINE_PUB_WAIT_MS);
+            props.setProperty(specialize(PoolingProperties.START_HEARTBEAT_MS, CONTROLLER1),
+                            "" + STD_START_HEARTBEAT_MS);
+            props.setProperty(specialize(PoolingProperties.REACTIVATE_MS, CONTROLLER1), "" + STD_REACTIVATE_WAIT_MS);
+            props.setProperty(specialize(PoolingProperties.IDENTIFICATION_MS, CONTROLLER1), "" + STD_IDENTIFICATION_MS);
+            props.setProperty(specialize(PoolingProperties.ACTIVE_HEARTBEAT_MS, CONTROLLER1),
+                            "" + STD_ACTIVE_HEARTBEAT_MS);
+            props.setProperty(specialize(PoolingProperties.INTER_HEARTBEAT_MS, CONTROLLER1),
+                            "" + STD_INTER_HEARTBEAT_MS);
+
+            props.putAll(makeSinkProperties(INTERNAL_TOPIC));
+            props.putAll(makeSourceProperties(INTERNAL_TOPIC));
+
+            return props;
+        }
+
+        @Override
+        public PolicyController getController(DroolsController droolsController) {
+            return context.getController(droolsController);
+        }
+    }
+
+    /**
+     * Simulator for the pooling manager factory.
+     */
+    private static class ManagerFactory extends PoolingManagerImpl.Factory {
+
+        /**
+         * Used to decode events from the external topic.
+         */
+        private final ThreadLocal<ObjectMapper> mapper = new ThreadLocal<ObjectMapper>() {
+            @Override
+            protected ObjectMapper initialValue() {
+                return new ObjectMapper();
+            }
+        };
+
+        /**
+         * Used to decode events into a Map.
+         */
+        private final TypeReference<TreeMap<String, String>> typeRef = new TypeReference<TreeMap<String, String>>() {};
+
+        /**
+         * 
+         * @param context
+         */
+        public ManagerFactory(Context context) {
+
+            /*
+             * Note: do NOT extract anything from "context" at this point, because it
+             * hasn't been fully initialized yet
+             */
+        }
+
+        @Override
+        public boolean canDecodeEvent(DroolsController drools, String topic) {
+            return true;
+        }
+
+        @Override
+        public Object decodeEvent(DroolsController drools, String topic, String event) {
+            try {
+                return mapper.get().readValue(event, typeRef);
+
+            } catch (IOException e) {
+                logger.warn("cannot decode external event", e);
+                return null;
+            }
+        }
+    }
+
+    /**
+     * Controller that also implements the {@link TopicListener} interface.
+     */
+    private static interface ListenerController extends PolicyController, TopicListener {
+
+    }
+}
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureTest.java
index 7782e475..9ee2d976 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureTest.java
@@ -22,6 +22,7 @@ package org.onap.policy.drools.pooling;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
@@ -122,8 +123,8 @@ public class PoolingFeatureTest {
         when(factory.getController(drools2)).thenReturn(controller2);
         when(factory.getController(droolsDisabled)).thenReturn(controllerDisabled);
 
-        when(factory.makeManager(any(), any())).thenAnswer(args -> {
-            PoolingProperties props = args.getArgument(1);
+        when(factory.makeManager(any(), any(), any())).thenAnswer(args -> {
+            PoolingProperties props = args.getArgument(2);
 
             PoolingManagerImpl mgr = mock(PoolingManagerImpl.class);
 
@@ -149,6 +150,19 @@ public class PoolingFeatureTest {
     }
 
     @Test
+    public void testGetHost() {
+        String host = pool.getHost();
+        assertNotNull(host);
+
+        // create another and ensure it generates another host name
+        pool = new PoolingFeature();
+        String host2 = pool.getHost();
+        assertNotNull(host2);
+
+        assertTrue(!host.equals(host2));
+    }
+
+    @Test
     public void testGetSequenceNumber() {
         assertEquals(0, pool.getSequenceNumber());
     }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java
index e32fa545..693cb6de 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java
@@ -23,6 +23,7 @@ package org.onap.policy.drools.pooling;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -73,6 +74,7 @@ public class PoolingManagerImplTest {
     protected static final long STD_INTER_HEARTBEAT_MS = STD_ACTIVE_HEARTBEAT_MS + 1;
     protected static final long STD_OFFLINE_PUB_WAIT_MS = STD_INTER_HEARTBEAT_MS + 1;
 
+    private static final String MY_HOST = "my.host";
     private static final String HOST2 = "other.host";
 
     private static final String MY_CONTROLLER = "my.controller";
@@ -151,7 +153,7 @@ public class PoolingManagerImplTest {
 
         when(factory.makeEventQueue(any())).thenReturn(eventQueue);
         when(factory.makeClassExtractors(any())).thenReturn(extractors);
-        when(factory.makeDmaapManager(any())).thenReturn(dmaap);
+        when(factory.makeDmaapManager(any(), any())).thenReturn(dmaap);
         when(factory.makeScheduler()).thenReturn(sched);
         when(factory.canDecodeEvent(drools, TOPIC2)).thenReturn(true);
         when(factory.decodeEvent(drools, TOPIC2, THE_EVENT)).thenReturn(DECODED_EVENT);
@@ -179,12 +181,12 @@ public class PoolingManagerImplTest {
 
         PoolingManagerImpl.setFactory(factory);
 
-        mgr = new PoolingManagerImpl(controller, poolProps);
+        mgr = new PoolingManagerImpl(MY_HOST, controller, poolProps);
     }
 
     @Test
-    public void testPoolingManagerImpl() {
-        mgr = new PoolingManagerImpl(controller, poolProps);
+    public void testPoolingManagerImpl() throws Exception {
+        verify(factory).makeDmaapManager(any(), any());
 
         State st = mgr.getCurrent();
         assertTrue(st instanceof IdleState);
@@ -202,7 +204,7 @@ public class PoolingManagerImplTest {
         PolicyController ctlr = mock(PolicyController.class);
 
         PoolingFeatureRtException ex = expectException(PoolingFeatureRtException.class,
-                        xxx -> new PoolingManagerImpl(ctlr, poolProps));
+                        () -> new PoolingManagerImpl(MY_HOST, ctlr, poolProps));
         assertNotNull(ex.getCause());
         assertTrue(ex.getCause() instanceof ClassCastException);
     }
@@ -211,23 +213,28 @@ public class PoolingManagerImplTest {
     public void testPoolingManagerImpl_PoolEx() throws PoolingFeatureException {
         // throw an exception when we try to create the dmaap manager
         PoolingFeatureException ex = new PoolingFeatureException();
-        when(factory.makeDmaapManager(any())).thenThrow(ex);
+        when(factory.makeDmaapManager(any(), any())).thenThrow(ex);
 
         PoolingFeatureRtException ex2 = expectException(PoolingFeatureRtException.class,
-                        xxx -> new PoolingManagerImpl(controller, poolProps));
+                        () -> new PoolingManagerImpl(MY_HOST, controller, poolProps));
         assertEquals(ex, ex2.getCause());
     }
 
     @Test
-    public void testGetHost() {
-        String host = mgr.getHost();
-        assertNotNull(host);
+    public void testGetCurrent() throws Exception {
+        assertEquals(IdleState.class, mgr.getCurrent().getClass());
+
+        startMgr();
+
+        assertEquals(StartState.class, mgr.getCurrent().getClass());
+    }
 
-        // create another manager and ensure it generates a different host
-        mgr = new PoolingManagerImpl(controller, poolProps);
+    @Test
+    public void testGetHost() {
+        assertEquals(MY_HOST, mgr.getHost());
 
-        assertNotNull(mgr.getHost());
-        assertFalse(host.equals(mgr.getHost()));
+        mgr = new PoolingManagerImpl(HOST2, controller, poolProps);
+        assertEquals(HOST2, mgr.getHost());
     }
 
     @Test
@@ -268,7 +275,7 @@ public class PoolingManagerImplTest {
         PoolingFeatureException ex = new PoolingFeatureException();
         doThrow(ex).when(dmaap).startPublisher();
 
-        PoolingFeatureException ex2 = expectException(PoolingFeatureException.class, xxx -> mgr.beforeStart());
+        PoolingFeatureException ex2 = expectException(PoolingFeatureException.class, () -> mgr.beforeStart());
         assertEquals(ex, ex2);
 
         // should never start the scheduler
@@ -453,8 +460,9 @@ public class PoolingManagerImplTest {
         // should have set the new filter
         verify(dmaap, times(++ntimes)).setFilter(any());
 
-        // should have cancelled the timer
-        assertEquals(1, futures.size());
+        // should have cancelled the timers
+        assertEquals(2, futures.size());
+        verify(futures.poll()).cancel(false);
         verify(futures.poll()).cancel(false);
 
         /*
@@ -465,8 +473,9 @@ public class PoolingManagerImplTest {
         // should have set the new filter
         verify(dmaap, times(++ntimes)).setFilter(any());
 
-        // timer should still be active
-        assertEquals(1, futures.size());
+        // new timers should now be active
+        assertEquals(2, futures.size());
+        verify(futures.poll(), never()).cancel(false);
         verify(futures.poll(), never()).cancel(false);
     }
 
@@ -548,7 +557,7 @@ public class PoolingManagerImplTest {
         ArgumentCaptor<Long> timeCap = ArgumentCaptor.forClass(Long.class);
         ArgumentCaptor<TimeUnit> unitCap = ArgumentCaptor.forClass(TimeUnit.class);
 
-        verify(sched).scheduleWithFixedDelay(taskCap.capture(), initCap.capture(), timeCap.capture(),
+        verify(sched, times(2)).scheduleWithFixedDelay(taskCap.capture(), initCap.capture(), timeCap.capture(),
                         unitCap.capture());
 
         assertEquals(STD_HEARTBEAT_WAIT_MS, initCap.getValue().longValue());
@@ -987,7 +996,7 @@ public class PoolingManagerImplTest {
 
         // route the message to this host
         mgr.startDistributing(makeAssignments(true));
-        
+
         // generate RuntimeException when onTopicEvent() is invoked
         doThrow(new IllegalArgumentException("expected")).when(controller).onTopicEvent(any(), any(), any());
 
@@ -1056,21 +1065,27 @@ public class PoolingManagerImplTest {
         startMgr();
 
         // route the message to this host
-        mgr.startDistributing(makeAssignments(true));
-
+        assertNotNull(mgr.startDistributing(makeAssignments(true)));
         assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
+        verify(eventQueue, never()).add(any());
 
 
-        // null assignments should be ignored
-        mgr.startDistributing(null);
+        // null assignments should cause message to be queued
+        assertNull(mgr.startDistributing(null));
+        assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
+        verify(eventQueue).add(any());
+
 
+        // route the message to this host
+        assertNotNull(mgr.startDistributing(makeAssignments(true)));
         assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
+        verify(eventQueue).add(any());
 
 
         // route the message to the other host
-        mgr.startDistributing(makeAssignments(false));
-
+        assertNotNull(mgr.startDistributing(makeAssignments(false)));
         assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
+        verify(eventQueue).add(any());
     }
 
     @Test
@@ -1086,7 +1101,9 @@ public class PoolingManagerImplTest {
         when(eventQueue.poll()).thenAnswer(args -> lst.poll());
 
         // route the messages to this host
-        assertTrue(mgr.startDistributing(makeAssignments(true)).await(2, TimeUnit.SECONDS));
+        CountDownLatch latch = mgr.startDistributing(makeAssignments(true));
+        assertNotNull(latch);
+        assertTrue(latch.await(2, TimeUnit.SECONDS));
 
         // all of the events should have been processed locally
         verify(dmaap, times(START_PUB)).publish(any());
@@ -1330,7 +1347,7 @@ public class PoolingManagerImplTest {
      */
     private <T extends Exception> T expectException(Class<T> exClass, ExFunction<T> func) {
         try {
-            func.apply(null);
+            func.apply();
             throw new AssertionError("missing exception");
 
         } catch (Exception e) {
@@ -1349,10 +1366,9 @@ public class PoolingManagerImplTest {
         /**
          * Invokes the function.
          * 
-         * @param arg always {@code null}
          * @throws T if an error occurs
          */
-        public void apply(Void arg) throws T;
+        public void apply() throws T;
 
     }
 }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingPropertiesTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingPropertiesTest.java
index 2d734c1c..459c770a 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingPropertiesTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingPropertiesTest.java
@@ -48,13 +48,13 @@ public class PoolingPropertiesTest {
     public static final boolean STD_FEATURE_ENABLED = true;
     public static final int STD_OFFLINE_LIMIT = 10;
     public static final long STD_OFFLINE_AGE_MS = 1000L;
-    public static final long STD_START_HEARTBEAT_MS = 2000L;
-    public static final long STD_REACTIVATE_MS = 3000L;
-    public static final long STD_IDENTIFICATION_MS = 4000L;
-    public static final long STD_LEADER_MS = 5000L;
-    public static final long STD_ACTIVE_HEARTBEAT_MS = 6000L;
-    public static final long STD_INTER_HEARTBEAT_MS = 7000L;
-    public static final long STD_OFFLINE_PUB_WAIT_MS = 8000L;
+    public static final long STD_OFFLINE_PUB_WAIT_MS = 2000L;
+    public static final long STD_START_HEARTBEAT_MS = 3000L;
+    public static final long STD_REACTIVATE_MS = 4000L;
+    public static final long STD_IDENTIFICATION_MS = 5000L;
+    public static final long STD_LEADER_MS = 6000L;
+    public static final long STD_ACTIVE_HEARTBEAT_MS = 7000L;
+    public static final long STD_INTER_HEARTBEAT_MS = 8000L;
 
     private Properties plain;
     private PoolingProperties pooling;
@@ -99,8 +99,13 @@ public class PoolingPropertiesTest {
     }
 
     @Test
+    public void testGetOfflinePubWaitMs() throws PropertyException {
+        doTest(OFFLINE_PUB_WAIT_MS, STD_OFFLINE_PUB_WAIT_MS, 3000L, xxx -> pooling.getOfflinePubWaitMs());
+    }
+
+    @Test
     public void testGetStartHeartbeatMs() throws PropertyException {
-        doTest(START_HEARTBEAT_MS, STD_START_HEARTBEAT_MS, 50000L, xxx -> pooling.getStartHeartbeatMs());
+        doTest(START_HEARTBEAT_MS, STD_START_HEARTBEAT_MS, 100000L, xxx -> pooling.getStartHeartbeatMs());
     }
 
     @Test
@@ -123,11 +128,6 @@ public class PoolingPropertiesTest {
         doTest(INTER_HEARTBEAT_MS, STD_INTER_HEARTBEAT_MS, 15000L, xxx -> pooling.getInterHeartbeatMs());
     }
 
-    @Test
-    public void testGetOfflinePubWaitMs() throws PropertyException {
-        doTest(OFFLINE_PUB_WAIT_MS, STD_OFFLINE_PUB_WAIT_MS, 3000L, xxx -> pooling.getOfflinePubWaitMs());
-    }
-
     /**
      * Tests a particular property. Verifies that the correct value is returned if the
      * specialized property has a value or the property has no value. Also verifies that
@@ -174,12 +174,12 @@ public class PoolingPropertiesTest {
         props.setProperty(specialize(FEATURE_ENABLED, CONTROLLER), "" + STD_FEATURE_ENABLED);
         props.setProperty(specialize(OFFLINE_LIMIT, CONTROLLER), "" + STD_OFFLINE_LIMIT);
         props.setProperty(specialize(OFFLINE_AGE_MS, CONTROLLER), "" + STD_OFFLINE_AGE_MS);
+        props.setProperty(specialize(OFFLINE_PUB_WAIT_MS, CONTROLLER), "" + STD_OFFLINE_PUB_WAIT_MS);
         props.setProperty(specialize(START_HEARTBEAT_MS, CONTROLLER), "" + STD_START_HEARTBEAT_MS);
         props.setProperty(specialize(REACTIVATE_MS, CONTROLLER), "" + STD_REACTIVATE_MS);
         props.setProperty(specialize(IDENTIFICATION_MS, CONTROLLER), "" + STD_IDENTIFICATION_MS);
         props.setProperty(specialize(ACTIVE_HEARTBEAT_MS, CONTROLLER), "" + STD_ACTIVE_HEARTBEAT_MS);
         props.setProperty(specialize(INTER_HEARTBEAT_MS, CONTROLLER), "" + STD_INTER_HEARTBEAT_MS);
-        props.setProperty(specialize(OFFLINE_PUB_WAIT_MS, CONTROLLER), "" + STD_OFFLINE_PUB_WAIT_MS);
 
         return props;
     }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/SpecPropertiesTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/SpecPropertiesTest.java
index 505dc400..d09650db 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/SpecPropertiesTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/SpecPropertiesTest.java
@@ -123,6 +123,24 @@ public class SpecPropertiesTest {
     }
 
     @Test
+    public void testSpecPropertiesStringStringProperties_EmptyPrefix() {
+        supportingProps = new Properties();
+
+        supportingProps.setProperty(PROP_NO_PREFIX, VAL_NO_PREFIX);
+        supportingProps.setProperty("a.value", VAL_GEN);
+        supportingProps.setProperty("b.value", VAL_GEN);
+        supportingProps.setProperty(MY_SPEC + ".b.value", VAL_SPEC);
+
+        // no supporting properties
+        props = new SpecProperties("", MY_SPEC, supportingProps);
+
+        assertEquals(VAL_NO_PREFIX, props.getProperty(gen(PROP_NO_PREFIX)));
+        assertEquals(VAL_GEN, props.getProperty(gen("a.value")));
+        assertEquals(VAL_SPEC, props.getProperty(MY_SPEC + ".b.value"));
+        assertNull(props.getProperty(gen(PROP_UNKNOWN)));
+    }
+
+    @Test
     public void testWithTrailingDot() {
         // neither has trailing dot
         assertEquals(PREFIX_GEN, props.getPrefix());
@@ -132,6 +150,16 @@ public class SpecPropertiesTest {
         props = new SpecProperties(PREFIX_GEN, MY_SPEC + ".");
         assertEquals(PREFIX_GEN, props.getPrefix());
         assertEquals(PREFIX_SPEC, props.getSpecPrefix());
+
+        // first is empty
+        props = new SpecProperties("", MY_SPEC);
+        assertEquals("", props.getPrefix());
+        assertEquals(MY_SPEC + ".", props.getSpecPrefix());
+
+        // second is empty
+        props = new SpecProperties(PREFIX_GEN, "");
+        assertEquals(PREFIX_GEN, props.getPrefix());
+        assertEquals(PREFIX_GEN, props.getSpecPrefix());
     }
 
     @Test
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ActiveStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ActiveStateTest.java
index 7b4b0602..27284dcd 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ActiveStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ActiveStateTest.java
@@ -298,18 +298,18 @@ public class ActiveStateTest extends BasicStateTester {
 
         // heart beat generator
         timer = repeatedTasks.remove();
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
 
         // my heart beat checker
         timer = repeatedTasks.remove();
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.first().longValue());
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second().longValue());
 
         // predecessor's heart beat checker
         timer = repeatedTasks.remove();
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.first().longValue());
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second().longValue());
     }
 
     @Test
@@ -327,13 +327,13 @@ public class ActiveStateTest extends BasicStateTester {
 
         // heart beat generator
         timer = repeatedTasks.remove();
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
 
         // my heart beat checker
         timer = repeatedTasks.remove();
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.first().longValue());
-        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second().longValue());
     }
 
     @Test
@@ -389,13 +389,13 @@ public class ActiveStateTest extends BasicStateTester {
 
         // set up next state
         State next = mock(State.class);
-        when(mgr.goInactive()).thenReturn(next);
+        when(mgr.goStart()).thenReturn(next);
 
         // fire the task - should transition
         assertEquals(next, task.third().fire());
 
-        // should indicate failure
-        verify(mgr).internalTopicFailed();
+        // should stop distributing
+        verify(mgr).startDistributing(null);
 
         // should publish an offline message
         Offline msg = captureAdminMessage(Offline.class);
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/InactiveStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/InactiveStateTest.java
index 394adaee..ae53ce05 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/InactiveStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/InactiveStateTest.java
@@ -22,12 +22,19 @@ package org.onap.policy.drools.pooling.state;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.mockito.ArgumentMatchers.any;
 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.Map;
 import org.junit.Before;
 import org.junit.Test;
+import org.onap.policy.drools.pooling.message.BucketAssignments;
+import org.onap.policy.drools.pooling.message.Identification;
+import org.onap.policy.drools.pooling.message.Leader;
 import org.onap.policy.drools.pooling.message.Message;
+import org.onap.policy.drools.pooling.message.Query;
 import org.onap.policy.drools.utils.Pair;
 
 public class InactiveStateTest extends BasicStateTester {
@@ -53,6 +60,42 @@ public class InactiveStateTest extends BasicStateTester {
     }
 
     @Test
+    public void testProcessLeader() {
+        State next = mock(State.class);
+        when(mgr.goActive()).thenReturn(next);
+
+        String[] arr = {PREV_HOST, MY_HOST, HOST1};
+        BucketAssignments asgn = new BucketAssignments(arr);
+        Leader msg = new Leader(PREV_HOST, asgn);
+
+        assertEquals(next, state.process(msg));
+        verify(mgr).startDistributing(asgn);
+    }
+
+    @Test
+    public void testProcessLeader_Invalid() {
+        Leader msg = new Leader(PREV_HOST, null);
+
+        // should stay in the same state, and not start distributing
+        assertNull(state.process(msg));
+        verify(mgr, never()).startDistributing(any());
+        verify(mgr, never()).goActive();
+        verify(mgr, never()).goInactive();
+    }
+
+    @Test
+    public void testProcessQuery() {
+        State next = mock(State.class);
+        when(mgr.goQuery()).thenReturn(next);
+
+        assertEquals(next, state.process(new Query()));
+
+        Identification ident = captureAdminMessage(Identification.class);
+        assertEquals(MY_HOST, ident.getSource());
+        assertEquals(ASGN3, ident.getAssignments());
+    }
+
+    @Test
     public void testGoInatcive() {
         assertNull(state.goInactive());
     }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ProcessingStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ProcessingStateTest.java
index e1718418..7ac58439 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ProcessingStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/ProcessingStateTest.java
@@ -65,39 +65,6 @@ public class ProcessingStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testGoActive_WithAssignment() {
-        State act = mock(State.class);
-        State inact = mock(State.class);
-
-        when(mgr.goActive()).thenReturn(act);
-        when(mgr.goInactive()).thenReturn(inact);
-
-        String[] arr = {HOST2, PREV_HOST, MY_HOST};
-        BucketAssignments asgn = new BucketAssignments(arr);
-
-        assertEquals(act, state.goActive(asgn));
-
-        verify(mgr).startDistributing(asgn);
-    }
-
-    @Test
-    public void testGoActive_WithoutAssignment() {
-        State act = mock(State.class);
-        State inact = mock(State.class);
-
-        when(mgr.goActive()).thenReturn(act);
-        when(mgr.goInactive()).thenReturn(inact);
-
-        String[] arr = {HOST2, PREV_HOST};
-        BucketAssignments asgn = new BucketAssignments(arr);
-
-        assertEquals(inact, state.goActive(asgn));
-
-        verify(mgr).startDistributing(asgn);
-
-    }
-
-    @Test
     public void testProcessQuery() {
         State next = mock(State.class);
         when(mgr.goQuery()).thenReturn(next);
@@ -155,13 +122,6 @@ public class ProcessingStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testMakeIdentification() {
-        Identification ident = state.makeIdentification();
-        assertEquals(MY_HOST, ident.getSource());
-        assertEquals(ASGN3, ident.getAssignments());
-    }
-
-    @Test
     public void testGetAssignments() {
         // assignments from constructor
         assertEquals(ASGN3, state.getAssignments());
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java
index a7c3a3d5..80778ed4 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java
@@ -21,7 +21,6 @@
 package org.onap.policy.drools.pooling.state;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
@@ -254,10 +253,13 @@ public class QueryStateTest extends BasicStateTester {
         // should published an Offline message and go inactive
 
         State next = mock(State.class);
-        when(mgr.goInactive()).thenReturn(next);
+        when(mgr.goStart()).thenReturn(next);
 
         assertEquals(next, timer.second().fire());
 
+        // should stop distributing
+        verify(mgr).startDistributing(null);
+
         Offline msg = captureAdminMessage(Offline.class);
         assertEquals(MY_HOST, msg.getSource());
     }
@@ -343,21 +345,6 @@ public class QueryStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testHasAssignment() {
-        // null assignment
-        mgr.startDistributing(null);
-        assertFalse(state.hasAssignment());
-
-        // not in assignments
-        state.setAssignments(new BucketAssignments(new String[] {HOST3}));
-        assertFalse(state.hasAssignment());
-
-        // it IS in the assignments
-        state.setAssignments(new BucketAssignments(new String[] {MY_HOST}));
-        assertTrue(state.hasAssignment());
-    }
-
-    @Test
     public void testRecordInfo_NullSource() {
         state.setAssignments(ASGN3);
         state.setLeader(MY_HOST);
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StartStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StartStateTest.java
index af4e8f13..ee4c1add 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StartStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StartStateTest.java
@@ -26,6 +26,7 @@ import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import java.util.Map;
@@ -39,6 +40,7 @@ import org.onap.policy.drools.pooling.message.Message;
 import org.onap.policy.drools.pooling.message.Offline;
 import org.onap.policy.drools.pooling.message.Query;
 import org.onap.policy.drools.utils.Pair;
+import org.onap.policy.drools.utils.Triple;
 
 public class StartStateTest extends BasicStateTester {
 
@@ -78,15 +80,36 @@ public class StartStateTest extends BasicStateTester {
         assertEquals(MY_HOST, msg.first());
         assertEquals(state.getHbTimestampMs(), msg.second().getTimestampMs());
 
-        Pair<Long, StateTimerTask> timer = onceTasks.removeFirst();
 
-        assertEquals(STD_HEARTBEAT_WAIT_MS, timer.first().longValue());
+        /*
+         * Verify heartbeat generator
+         */
+        Triple<Long, Long, StateTimerTask> generator = repeatedTasks.removeFirst();
+
+        assertEquals(STD_INTER_HEARTBEAT_MS, generator.first().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, generator.second().longValue());
+
+        // invoke the task - it should generate another heartbeat
+        assertEquals(null, generator.third().fire());
+        verify(mgr, times(2)).publish(MY_HOST, msg.second());
+
+        // and again
+        assertEquals(null, generator.third().fire());
+        verify(mgr, times(3)).publish(MY_HOST, msg.second());
+
+
+        /*
+         * Verify heartbeat checker
+         */
+        Pair<Long, StateTimerTask> checker = onceTasks.removeFirst();
+
+        assertEquals(STD_HEARTBEAT_WAIT_MS, checker.first().longValue());
 
         // invoke the task - it should go to the state returned by the mgr
         State next = mock(State.class);
         when(mgr.goInactive()).thenReturn(next);
 
-        assertEquals(next, timer.second().fire());
+        assertEquals(next, checker.second().fire());
 
         verify(mgr).internalTopicFailed();
     }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StateTest.java
index a184dfad..08b55c6b 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/StateTest.java
@@ -154,12 +154,35 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
-    public void testGoActive() {
-        State next = mock(State.class);
-        when(mgr.goActive()).thenReturn(next);
+    public void testGoActive_WithAssignment() {
+        State act = mock(State.class);
+        State inact = mock(State.class);
 
-        State next2 = state.goActive();
-        assertEquals(next, next2);
+        when(mgr.goActive()).thenReturn(act);
+        when(mgr.goInactive()).thenReturn(inact);
+
+        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
+
+        assertEquals(act, state.goActive(asgn));
+
+        verify(mgr).startDistributing(asgn);
+    }
+
+    @Test
+    public void testGoActive_WithoutAssignment() {
+        State act = mock(State.class);
+        State inact = mock(State.class);
+
+        when(mgr.goActive()).thenReturn(act);
+        when(mgr.goInactive()).thenReturn(inact);
+
+        String[] arr = {HOST2, PREV_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
+
+        assertEquals(inact, state.goActive(asgn));
+
+        verify(mgr).startDistributing(asgn);
     }
 
     @Test
@@ -375,6 +398,20 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
+    public void testMissedHeartbeat() {
+        State next = mock(State.class);
+        when(mgr.goStart()).thenReturn(next);
+
+        State next2 = state.missedHeartbeat();
+        assertEquals(next, next2);
+
+        verify(mgr).startDistributing(null);
+
+        Offline msg = captureAdminMessage(Offline.class);
+        assertEquals(MY_HOST, msg.getSource());
+    }
+
+    @Test
     public void testInternalTopicFailed() {
         State next = mock(State.class);
         when(mgr.goInactive()).thenReturn(next);
@@ -398,6 +435,13 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
+    public void testMakeIdentification() {
+        Identification ident = state.makeIdentification();
+        assertEquals(MY_HOST, ident.getSource());
+        assertEquals(ASGN3, ident.getAssignments());
+    }
+
+    @Test
     public void testMakeOffline() {
         Offline msg = state.makeOffline();