Merge "Sonar fixes to pooling"

18 files changed, 1737 insertions, 498 deletions

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 f68f2395..a91671fd 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
@@ -21,6 +21,8 @@
 package org.onap.policy.drools.pooling;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doThrow;
@@ -211,11 +213,11 @@ public class DmaapManagerTest {
 
         expectException("startPublisher,start", xxx -> mgr.startPublisher());
         expectException("startPublisher,publish", xxx -> mgr.publish(MSG));
-        
+
         // allow it to succeed this time
         reset(sink);
         when(sink.send(any())).thenReturn(true);
-        
+
         mgr.startPublisher();
         verify(sink).start();
 
@@ -227,29 +229,64 @@ public class DmaapManagerTest {
     @Test
     public void testStopPublisher() throws PoolingFeatureException {
         // not publishing yet, so stopping should have no effect
-        mgr.stopPublisher();
+        mgr.stopPublisher(0);
         verify(sink, never()).stop();
-        
+
         // now start it
         mgr.startPublisher();
-        
+
         // this time, stop should do something
-        mgr.stopPublisher();
+        mgr.stopPublisher(0);
         verify(sink).stop();
-        
+
         // re-stopping should have no effect
-        mgr.stopPublisher();
+        mgr.stopPublisher(0);
         verify(sink).stop();
     }
 
     @Test
+    public void testStopPublisher_WithDelay() throws PoolingFeatureException {
+
+        mgr.startPublisher();
+
+        long tbeg = System.currentTimeMillis();
+
+        mgr.stopPublisher(100L);
+
+        assertTrue(System.currentTimeMillis() >= tbeg + 100L);
+    }
+
+    @Test
+    public void testStopPublisher_WithDelayInterrupted() throws Exception {
+
+        mgr.startPublisher();
+
+        long minms = 2000L;
+
+        // tell the publisher to stop in minms + additional time
+        Thread thread = new Thread(() -> mgr.stopPublisher(minms + 3000L));
+        thread.start();
+
+        // give the thread a chance to start
+        Thread.sleep(50L);
+
+        // interrupt it - it should immediately finish its work
+        thread.interrupt();
+
+        // wait for it to stop, but only wait the minimum time
+        thread.join(minms);
+
+        assertFalse(thread.isAlive());
+    }
+
+    @Test
     public void testStopPublisher_Exception() throws PoolingFeatureException {
         mgr.startPublisher();
-        
+
         // force exception when it stops
         doThrow(new IllegalStateException("expected")).when(sink).stop();
 
-        mgr.stopPublisher();
+        mgr.stopPublisher(0);
     }
 
     @Test
@@ -270,14 +307,14 @@ public class DmaapManagerTest {
         // not consuming yet, so stopping should have no effect
         mgr.stopConsumer(listener);
         verify(source, never()).unregister(any());
-        
+
         // now start it
         mgr.startConsumer(listener);
-        
+
         // this time, stop should do something
         mgr.stopConsumer(listener);
         verify(source).unregister(listener);
-        
+
         // re-stopping should have no effect
         mgr.stopConsumer(listener);
         verify(source).unregister(listener);
@@ -292,7 +329,7 @@ public class DmaapManagerTest {
     public void testSetFilter_Exception() throws PoolingFeatureException {
         // force an error when setFilter() is called
         doThrow(new UnsupportedOperationException("expected")).when(source).setFilter(any());
-        
+
         mgr.setFilter(FILTER);
     }
 
@@ -300,41 +337,41 @@ public class DmaapManagerTest {
     public void testPublish() throws PoolingFeatureException {
         // cannot publish before starting
         expectException("publish,pre", xxx -> mgr.publish(MSG));
-        
+
         mgr.startPublisher();
-        
+
         // publish several messages
         mgr.publish(MSG);
         verify(sink).send(MSG);
-        
-        mgr.publish(MSG+"a");
-        verify(sink).send(MSG+"a");
-        
-        mgr.publish(MSG+"b");
-        verify(sink).send(MSG+"b");
-        
+
+        mgr.publish(MSG + "a");
+        verify(sink).send(MSG + "a");
+
+        mgr.publish(MSG + "b");
+        verify(sink).send(MSG + "b");
+
         // stop and verify we can no longer publish
-        mgr.stopPublisher();
+        mgr.stopPublisher(0);
         expectException("publish,stopped", xxx -> mgr.publish(MSG));
     }
 
     @Test(expected = PoolingFeatureException.class)
     public void testPublish_SendFailed() throws PoolingFeatureException {
         mgr.startPublisher();
-        
+
         // arrange for send() to fail
         when(sink.send(MSG)).thenReturn(false);
-        
+
         mgr.publish(MSG);
     }
 
     @Test(expected = PoolingFeatureException.class)
     public void testPublish_SendEx() throws PoolingFeatureException {
         mgr.startPublisher();
-        
+
         // arrange for send() to throw an exception
         doThrow(new IllegalStateException("expected")).when(sink).send(MSG);
-        
+
         mgr.publish(MSG);
     }
 
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureEnabledCheckerTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureEnabledCheckerTest.java
deleted file mode 100644
index 5f918f73..00000000
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureEnabledCheckerTest.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * ============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.assertFalse;
-import static org.junit.Assert.assertTrue;
-import static org.onap.policy.common.utils.properties.SpecPropertyConfiguration.generalize;
-import static org.onap.policy.common.utils.properties.SpecPropertyConfiguration.specialize;
-import java.util.Properties;
-import org.junit.Test;
-
-public class FeatureEnabledCheckerTest {
-
-    private static final String PROP_NAME = "enable.{?.}it";
-
-    private static final String SPEC = "my.specializer";
-
-    @Test
-    public void test() {
-        assertFalse(check(null, null));
-        assertTrue(check(null, true));
-        assertFalse(check(null, false));
-
-        assertTrue(check(true, null));
-        assertTrue(check(true, true));
-        assertFalse(check(true, false));
-
-        assertFalse(check(false, null));
-        assertTrue(check(false, true));
-        assertFalse(check(false, false));
-    }
-
-    /**
-     * Adds properties, as specified, and checks if the feature is enabled.
-     * 
-     * @param wantGen value to assign to the generalized property, or
-     *        {@code null} to leave it unset
-     * @param wantSpec value to assign to the specialized property, or
-     *        {@code null} to leave it unset
-     * @return {@code true} if the feature is enabled, {@code false} otherwise
-     */
-    public boolean check(Boolean wantGen, Boolean wantSpec) {
-        Properties props = new Properties();
-
-        if (wantGen != null) {
-            props.setProperty(generalize(PROP_NAME), wantGen.toString());
-        }
-
-        if (wantSpec != null) {
-            props.setProperty(specialize(PROP_NAME, SPEC), wantSpec.toString());
-        }
-
-        return FeatureEnabledChecker.isFeatureEnabled(props, SPEC, PROP_NAME);
-    }
-
-}
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
new file mode 100644
index 00000000..13d70f52
--- /dev/null
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest.java
@@ -0,0 +1,1158 @@
+/*
+ * ============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 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;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+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;
+import org.onap.policy.drools.controller.DroolsController;
+import org.onap.policy.drools.event.comm.FilterableTopicSource;
+import org.onap.policy.drools.event.comm.Topic;
+import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
+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.pooling.message.Message;
+import org.onap.policy.drools.system.PolicyController;
+import org.onap.policy.drools.system.PolicyEngine;
+import org.onap.policy.drools.utils.Pair;
+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.
+ */
+public class FeatureTest {
+
+    private static final Logger logger = LoggerFactory.getLogger(FeatureTest.class);
+
+    /**
+     * Name of the topic used for inter-host communication.
+     */
+    private static final String INTERNAL_TOPIC = "my.internal.topic";
+
+    /**
+     * Name of the topic from which "external" events "arrive".
+     */
+    private static final String EXTERNAL_TOPIC = "my.external.topic";
+
+    /**
+     * Name of the controller.
+     */
+    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;
+
+    // these are saved and restored on exit from this test class
+    private static PoolingFeature.Factory saveFeatureFactory;
+    private static PoolingManagerImpl.Factory saveManagerFactory;
+    private static DmaapManager.Factory saveDmaapFactory;
+
+    /**
+     * Context for the current test case.
+     */
+    private Context ctx;
+
+    @BeforeClass
+    public static void setUpBeforeClass() {
+        saveFeatureFactory = PoolingFeature.getFactory();
+        saveManagerFactory = PoolingManagerImpl.getFactory();
+        saveDmaapFactory = DmaapManager.getFactory();
+    }
+
+    @AfterClass
+    public static void tearDownAfterClass() {
+        PoolingFeature.setFactory(saveFeatureFactory);
+        PoolingManagerImpl.setFactory(saveManagerFactory);
+        DmaapManager.setFactory(saveDmaapFactory);
+    }
+
+    @Before
+    public void setUp() {
+        ctx = null;
+    }
+
+    @After
+    public void tearDown() {
+        if (ctx != null) {
+            ctx.destroy();
+        }
+    }
+
+    @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();
+    }
+
+    @Ignore
+    @Test
+    public void test_TwoHosts() throws Exception {
+        int nmessages = 200;
+
+        ctx = new Context(nmessages);
+
+        ctx.addHost();
+        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);
+
+        assertEquals(0, ctx.getDecodeErrors());
+        assertEquals(0, ctx.getRemainingEvents());
+        ctx.checkAllSawAMsg();
+    }
+
+    private String makeMessage(int reqnum) {
+        return "{\"reqid\":\"req" + reqnum + "\", \"data\":\"hello " + reqnum + "\"}";
+    }
+
+    /**
+     * Context used for a single test case.
+     */
+    private static class Context {
+
+        private final FeatureFactory featureFactory;
+        private final ManagerFactory managerFactory;
+        private final DmaapFactory dmaapFactory;
+
+        /**
+         * 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<>();
+
+        /**
+         * Maps a channel to its queue. Does <i>not</i> include the "admin" channel.
+         */
+        private final ConcurrentMap<String, BlockingQueue<String>> channel2queue = new ConcurrentHashMap<>(7);
+
+        /**
+         * Queue for the external "DMaaP" topic.
+         */
+        private final BlockingQueue<String> externalTopic = new LinkedBlockingQueue<String>();
+
+        /**
+         * 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;
+
+        /**
+         * 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()}.
+         */
+        private Host currentHost = null;
+
+        /**
+         * 
+         * @param nEvents number of events to be processed
+         */
+        public Context(int nEvents) {
+            featureFactory = new FeatureFactory(this);
+            managerFactory = new ManagerFactory(this);
+            dmaapFactory = new DmaapFactory(this);
+            eventCounter = new CountDownLatch(nEvents);
+
+            PoolingFeature.setFactory(featureFactory);
+            PoolingManagerImpl.setFactory(managerFactory);
+            DmaapManager.setFactory(dmaapFactory);
+        }
+
+        /**
+         * Destroys the context, stopping any hosts that remain.
+         */
+        public void destroy() {
+            stopHosts();
+            hosts.clear();
+        }
+
+        /**
+         * Creates and adds a new host to the context.
+         */
+        public void addHost() {
+            hosts.add(new Host(this));
+        }
+
+        /**
+         * 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;
+            }
+        }
+
+        /**
+         * Sets {@link #currentHost} to the specified host, and then invokes the given
+         * function. Resets {@link #currentHost} to {@code null} before returning.
+         * 
+         * @param host
+         * @param func function to invoke
+         */
+        public void withHost(Host host, VoidFunction func) {
+            currentHost = host;
+            func.apply();
+            currentHost = null;
+        }
+
+        /**
+         * Offers an event to the external topic.
+         * 
+         * @param event
+         */
+        public void offerExternal(String event) {
+            externalTopic.offer(event);
+        }
+
+        /**
+         * Adds an internal channel to the set of channels.
+         * 
+         * @param channel
+         * @param queue the channel's queue
+         */
+        public void addInternal(String channel, BlockingQueue<String> queue) {
+            channel2queue.put(channel, queue);
+        }
+
+        /**
+         * Offers a message to all internal channels.
+         * 
+         * @param message
+         */
+        public void offerInternal(String message) {
+            channel2queue.values().forEach(queue -> queue.offer(message));
+        }
+
+        /**
+         * Offers amessage to an internal channel.
+         * 
+         * @param channel
+         * @param message
+         */
+        public void offerInternal(String channel, String message) {
+            BlockingQueue<String> queue = channel2queue.get(channel);
+            if (queue != null) {
+                queue.offer(message);
+            }
+        }
+
+        /**
+         * 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 queue for the external topic
+         */
+        public BlockingQueue<String> getExternalTopic() {
+            return externalTopic;
+        }
+
+        /**
+         * 
+         * @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);
+        }
+
+        /**
+         * 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)}.
+         * 
+         * @return the current host, or {@code null} if there is no current host
+         */
+        public Host getCurrentHost() {
+            return currentHost;
+        }
+    }
+
+    /**
+     * Simulates a single "host".
+     */
+    private static class Host {
+
+        private final Context context;
+
+        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);
+
+        /**
+         * This host's internal "DMaaP" topic.
+         */
+        private final BlockingQueue<String> msgQueue = new LinkedBlockingQueue<>();
+
+        /**
+         * Source that reads from the external topic and posts to the listener.
+         */
+        private 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) {
+            this.context = context;
+
+            when(controller.getName()).thenReturn(CONTROLLER1);
+            when(controller.getDrools()).thenReturn(drools);
+
+            // 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);
+
+            // arrange to read from the external topic
+            externalSource = new TopicSourceImpl(context, false);
+        }
+
+        /**
+         * Gets the host name. This should only be invoked within {@link #start()}.
+         * 
+         * @return the host name
+         */
+        public String getName() {
+            return PoolingManagerImpl.getLastHost();
+        }
+
+        /**
+         * 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() {
+
+            context.withHost(this, () -> {
+
+                feature.beforeStart(engine);
+                feature.afterCreate(controller);
+
+                // assign the queue for this host's internal topic
+                context.addInternal(getName(), msgQueue);
+
+                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();
+        }
+
+        /**
+         * @return the queue associated with this host's internal topic
+         */
+        public BlockingQueue<String> getInternalQueue() {
+            return msgQueue;
+        }
+    }
+
+    /**
+     * 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;
+        }
+    }
+
+    /**
+     * Sink implementation that puts a message on the queue specified by the
+     * <i>channel</i> embedded within the message. If it's the "admin" channel, then the
+     * message is placed on all queues.
+     */
+    private static class TopicSinkImpl extends TopicImpl implements TopicSink {
+
+        private final Context context;
+
+        /**
+         * Used to decode the messages so that the channel can be extracted.
+         */
+        private final Serializer serializer = new Serializer();
+
+        /**
+         * 
+         * @param context
+         */
+        public TopicSinkImpl(Context context) {
+            this.context = context;
+        }
+
+        @Override
+        public synchronized boolean send(String message) {
+            if (!isAlive()) {
+                return false;
+            }
+
+            try {
+                Message msg = serializer.decodeMsg(message);
+                String channel = msg.getChannel();
+
+                if (Message.ADMIN.equals(channel)) {
+                    // add to every queue
+                    context.offerInternal(message);
+
+                } else {
+                    // add to a specific queue
+                    context.offerInternal(channel, message);
+                }
+
+                return true;
+
+            } catch (IOException e) {
+                logger.warn("could not decode message: {}", message);
+                context.bumpDecodeErrors();
+                return false;
+            }
+        }
+    }
+
+    /**
+     * Source implementation that reads from a queue associated with a topic.
+     */
+    private static class TopicSourceImpl extends TopicImpl implements FilterableTopicSource {
+
+        private final String topic;
+
+        /**
+         * Queue from which to retrieve messages.
+         */
+        private final BlockingQueue<String> queue;
+
+        /**
+         * Manages the current consumer thread. The "first" item is used as a trigger to
+         * tell the thread to stop processing, while the "second" item is triggered <i>by
+         * the thread</i> when it completes.
+         */
+        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}
+         *        to read from the external topic
+         */
+        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);
+
+            } else {
+                this.topic = EXTERNAL_TOPIC;
+                this.queue = context.getExternalTopic();
+            }
+        }
+
+        @Override
+        public void setFilter(String filter) {
+            logger.info("topic filter set to: {}", filter);
+        }
+
+        @Override
+        public String getTopic() {
+            return topic;
+        }
+
+        @Override
+        public boolean offer(String event) {
+            throw new UnsupportedOperationException("offer topic source");
+        }
+
+        /**
+         * Starts a thread that takes messages from the queue and gives them to the
+         * listener. Stops the thread of any previously registered listener.
+         */
+        @Override
+        public void register(TopicListener listener) {
+            Pair<CountDownLatch, CountDownLatch> newPair = new Pair<>(new CountDownLatch(1), new CountDownLatch(1));
+
+            reregister(newPair);
+
+            new Thread(() -> {
+                try {
+                    do {
+                        processMessages(newPair.first(), listener);
+                    } while (!newPair.first().await(INTER_POLL_MS, TimeUnit.MILLISECONDS));
+
+                    logger.info("topic source thread completed");
+
+                } catch (InterruptedException e) {
+                    logger.warn("topic source thread aborted", e);
+                    Thread.currentThread().interrupt();
+
+                } catch (RuntimeException e) {
+                    logger.warn("topic source thread aborted", e);
+                }
+
+                newPair.second().countDown();
+
+            }).start();
+        }
+
+        /**
+         * Stops the thread of <i>any</i> currently registered listener.
+         */
+        @Override
+        public void unregister(TopicListener listener) {
+            reregister(null);
+        }
+
+        /**
+         * Registers a new "pair" with this source, stopping the consumer associated with
+         * any previous registration.
+         * 
+         * @param newPair the new "pair", or {@code null} to unregister
+         */
+        private void reregister(Pair<CountDownLatch, CountDownLatch> newPair) {
+            try {
+                Pair<CountDownLatch, CountDownLatch> oldPair = pair.getAndSet(newPair);
+                if (oldPair == null) {
+                    if (newPair == null) {
+                        // unregister was invoked twice in a row
+                        logger.warn("re-unregister for topic source");
+                    }
+
+                    // no previous thread to stop
+                    return;
+                }
+
+                // need to stop the previous thread
+
+                // tell it to stop
+                oldPair.first().countDown();
+
+                // wait for it to stop
+                if (!oldPair.second().await(2, TimeUnit.SECONDS)) {
+                    logger.warn("old topic registration is still running");
+                }
+
+            } catch (InterruptedException e) {
+                logger.warn("old topic registration may still be running", e);
+                Thread.currentThread().interrupt();
+            }
+
+            if (newPair != null) {
+                // register was invoked twice in a row
+                logger.warn("re-register for topic source");
+            }
+        }
+
+        /**
+         * 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.
+         * 
+         * @param stopped triggered if processing should stop
+         * @param listener
+         * @throws InterruptedException
+         */
+        private void processMessages(CountDownLatch stopped, TopicListener listener) throws InterruptedException {
+
+            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);
+                if (msg == null) {
+                    return;
+                }
+
+                listener.onTopicEvent(CommInfrastructure.UEB, topic, msg);
+            }
+        }
+    }
+
+    /**
+     * Topic implementation. Most methods just throw
+     * {@link UnsupportedOperationException}.
+     */
+    private static class TopicImpl implements Topic {
+
+        /**
+         * {@code True} if this topic is alive/running, {@code false} otherwise.
+         */
+        private boolean alive = false;
+
+        /**
+         * 
+         */
+        public TopicImpl() {
+            super();
+        }
+
+        @Override
+        public String getTopic() {
+            return INTERNAL_TOPIC;
+        }
+
+        @Override
+        public CommInfrastructure getTopicCommInfrastructure() {
+            throw new UnsupportedOperationException("topic protocol");
+        }
+
+        @Override
+        public List<String> getServers() {
+            throw new UnsupportedOperationException("topic servers");
+        }
+
+        @Override
+        public String[] getRecentEvents() {
+            throw new UnsupportedOperationException("topic events");
+        }
+
+        @Override
+        public void register(TopicListener topicListener) {
+            throw new UnsupportedOperationException("register topic");
+        }
+
+        @Override
+        public void unregister(TopicListener topicListener) {
+            throw new UnsupportedOperationException("unregister topic");
+        }
+
+        @Override
+        public synchronized boolean start() {
+            if (alive) {
+                throw new IllegalStateException("topic already started");
+            }
+
+            alive = true;
+            return true;
+        }
+
+        @Override
+        public synchronized boolean stop() {
+            if (!alive) {
+                throw new IllegalStateException("topic is not running");
+            }
+
+            alive = false;
+            return true;
+        }
+
+        @Override
+        public synchronized void shutdown() {
+            alive = false;
+        }
+
+        @Override
+        public synchronized boolean isAlive() {
+            return alive;
+        }
+
+        @Override
+        public boolean lock() {
+            throw new UnsupportedOperationException("lock topicink");
+        }
+
+        @Override
+        public boolean unlock() {
+            throw new UnsupportedOperationException("unlock topic");
+        }
+
+        @Override
+        public boolean isLocked() {
+            throw new UnsupportedOperationException("topic isLocked");
+        }
+    }
+
+    /**
+     * 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("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);
+
+            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;
+            }
+        }
+    }
+
+    /**
+     * Simulator for the dmaap manager factory.
+     */
+    private static class DmaapFactory extends DmaapManager.Factory {
+
+        private final Context context;
+
+        /**
+         * 
+         * @param context
+         */
+        public DmaapFactory(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 List<TopicSource> initTopicSources(Properties props) {
+            return Arrays.asList(new TopicSourceImpl(context, true));
+        }
+
+        @Override
+        public List<TopicSink> initTopicSinks(Properties props) {
+            return Arrays.asList(new TopicSinkImpl(context));
+        }
+    }
+
+    /**
+     * Controller that also implements the {@link TopicListener} interface.
+     */
+    private static interface ListenerController extends PolicyController, TopicListener {
+
+    }
+
+    /**
+     * Simple function that takes no arguments and returns nothing.
+     */
+    @FunctionalInterface
+    private static interface VoidFunction {
+
+        public void apply();
+    }
+}
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureExceptionTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureExceptionTest.java
index 5b423d4b..34b604c9 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureExceptionTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingFeatureExceptionTest.java
@@ -31,12 +31,4 @@ public class PoolingFeatureExceptionTest extends ExceptionsTester {
         assertEquals(5, test(PoolingFeatureException.class));
     }
 
-    @Test
-    public void testToRuntimeException() {
-        PoolingFeatureException plainExc = new PoolingFeatureException("hello");
-        PoolingFeatureRtException runtimeExc = plainExc.toRuntimeException();
-        
-        assertTrue(plainExc == runtimeExc.getCause());
-    }
-
 }
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 cd1aea09..7782e475 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
@@ -23,7 +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.assertTrue;
-import static org.mockito.Matchers.any;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -32,6 +32,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Properties;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -40,12 +41,11 @@ import org.onap.policy.drools.controller.DroolsController;
 import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.drools.pooling.PoolingFeature.Factory;
 import org.onap.policy.drools.system.PolicyController;
+import org.onap.policy.drools.system.PolicyEngine;
 import org.onap.policy.drools.utils.Pair;
 
 public class PoolingFeatureTest {
 
-    private static final String CONFIG_DIR = "src/test/java/org/onap/policy/drools/pooling";
-
     private static final String CONTROLLER1 = "controllerA";
     private static final String CONTROLLER2 = "controllerB";
     private static final String CONTROLLER_DISABLED = "controllerDisabled";
@@ -66,6 +66,8 @@ public class PoolingFeatureTest {
      */
     private static Factory saveFactory;
 
+    private Properties props;
+    private PolicyEngine engine;
     private PolicyController controller1;
     private PolicyController controller2;
     private PolicyController controllerDisabled;
@@ -94,6 +96,8 @@ public class PoolingFeatureTest {
 
     @Before
     public void setUp() throws Exception {
+        props = initProperties();
+        engine = mock(PolicyEngine.class);
         factory = mock(Factory.class);
         controller1 = mock(PolicyController.class);
         controller2 = mock(PolicyController.class);
@@ -113,12 +117,13 @@ public class PoolingFeatureTest {
         when(controllerException.getName()).thenReturn(CONTROLLER_EX);
         when(controllerUnknown.getName()).thenReturn(CONTROLLER_UNKNOWN);
 
+        when(factory.getProperties(PoolingProperties.FEATURE_NAME)).thenReturn(props);
         when(factory.getController(drools1)).thenReturn(controller1);
         when(factory.getController(drools2)).thenReturn(controller2);
         when(factory.getController(droolsDisabled)).thenReturn(controllerDisabled);
 
         when(factory.makeManager(any(), any())).thenAnswer(args -> {
-            PoolingProperties props = args.getArgumentAt(1, PoolingProperties.class);
+            PoolingProperties props = args.getArgument(1);
 
             PoolingManagerImpl mgr = mock(PoolingManagerImpl.class);
 
@@ -129,7 +134,7 @@ public class PoolingFeatureTest {
 
         pool = new PoolingFeature();
 
-        pool.globalInit(null, CONFIG_DIR);
+        pool.beforeStart(engine);
 
         pool.afterCreate(controller1);
         pool.afterCreate(controller2);
@@ -149,24 +154,17 @@ public class PoolingFeatureTest {
     }
 
     @Test
-    public void testGlobalInit() {
-        pool = new PoolingFeature();
-
-        pool.globalInit(null, CONFIG_DIR);
-    }
-
-    @Test(expected = PoolingFeatureRtException.class)
-    public void testGlobalInit_NotFound() {
+    public void testBeforeStartEngine() {
         pool = new PoolingFeature();
 
-        pool.globalInit(null, CONFIG_DIR + "/unknown");
+        assertFalse(pool.beforeStart(engine));
     }
 
     @Test
     public void testAfterCreate() {
         managers.clear();
         pool = new PoolingFeature();
-        pool.globalInit(null, CONFIG_DIR);
+        pool.beforeStart(engine);
 
         assertFalse(pool.afterCreate(controller1));
         assertEquals(1, managers.size());
@@ -184,7 +182,7 @@ public class PoolingFeatureTest {
     public void testAfterCreate_NotEnabled() {
         managers.clear();
         pool = new PoolingFeature();
-        pool.globalInit(null, CONFIG_DIR);
+        pool.beforeStart(engine);
 
         assertFalse(pool.afterCreate(controllerDisabled));
         assertTrue(managers.isEmpty());
@@ -194,7 +192,7 @@ public class PoolingFeatureTest {
     public void testAfterCreate_PropertyEx() {
         managers.clear();
         pool = new PoolingFeature();
-        pool.globalInit(null, CONFIG_DIR);
+        pool.beforeStart(engine);
 
         pool.afterCreate(controllerException);
     }
@@ -202,9 +200,9 @@ public class PoolingFeatureTest {
     @Test(expected = PoolingFeatureRtException.class)
     public void testAfterCreate_NoProps() {
         pool = new PoolingFeature();
-        
+
         // did not perform globalInit, which is an error
-        
+
         pool.afterCreate(controller1);
     }
 
@@ -212,7 +210,7 @@ public class PoolingFeatureTest {
     public void testAfterCreate_NoFeatProps() {
         managers.clear();
         pool = new PoolingFeature();
-        pool.globalInit(null, CONFIG_DIR);
+        pool.beforeStart(engine);
 
         assertFalse(pool.afterCreate(controllerUnknown));
         assertTrue(managers.isEmpty());
@@ -492,4 +490,31 @@ public class PoolingFeatureTest {
         pool.afterStop(controller1);
     }
 
+    private Properties initProperties() {
+        Properties props = new Properties();
+
+        initProperties(props, "A", 0);
+        initProperties(props, "B", 1);
+        initProperties(props, "Exception", 2);
+
+        props.setProperty("pooling.controllerDisabled.enabled", "false");
+
+        props.setProperty("pooling.controllerException.offline.queue.limit", "INVALID NUMBER");
+
+        return props;
+    }
+
+    private void initProperties(Properties props, String suffix, int offset) {
+        props.setProperty("pooling.controller" + suffix + ".topic", "topic." + suffix);
+        props.setProperty("pooling.controller" + suffix + ".enabled", "true");
+        props.setProperty("pooling.controller" + suffix + ".offline.queue.limit", String.valueOf(5 + offset));
+        props.setProperty("pooling.controller" + suffix + ".offline.queue.age.milliseconds",
+                        String.valueOf(100 + offset));
+        props.setProperty("pooling.controller" + suffix + ".start.heartbeat.milliseconds", String.valueOf(10 + offset));
+        props.setProperty("pooling.controller" + suffix + ".reactivate.milliseconds", String.valueOf(20 + offset));
+        props.setProperty("pooling.controller" + suffix + ".identification.milliseconds", String.valueOf(30 + offset));
+        props.setProperty("pooling.controller" + suffix + ".active.heartbeat.milliseconds",
+                        String.valueOf(40 + offset));
+        props.setProperty("pooling.controller" + suffix + ".inter.heartbeat.milliseconds", String.valueOf(50 + offset));
+    }
 }
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 01ee61ef..e32fa545 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
@@ -25,7 +25,8 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import static org.mockito.Matchers.any;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.contains;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -40,7 +41,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -71,6 +71,7 @@ public class PoolingManagerImplTest {
     protected static final long STD_IDENTIFICATION_MS = STD_REACTIVATE_WAIT_MS + 1;
     protected static final long STD_ACTIVE_HEARTBEAT_MS = STD_IDENTIFICATION_MS + 1;
     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 HOST2 = "other.host";
 
@@ -85,8 +86,8 @@ public class PoolingManagerImplTest {
     private static final String REQUEST_ID = "my.request.id";
 
     /**
-     * Number of dmaap.publish() invocations that should be issued when the
-     * manager is started.
+     * Number of dmaap.publish() invocations that should be issued when the manager is
+     * started.
      */
     private static final int START_PUB = 1;
 
@@ -135,6 +136,7 @@ public class PoolingManagerImplTest {
         when(poolProps.getIdentificationMs()).thenReturn(STD_IDENTIFICATION_MS);
         when(poolProps.getActiveHeartbeatMs()).thenReturn(STD_ACTIVE_HEARTBEAT_MS);
         when(poolProps.getInterHeartbeatMs()).thenReturn(STD_INTER_HEARTBEAT_MS);
+        when(poolProps.getOfflinePubWaitMs()).thenReturn(STD_OFFLINE_PUB_WAIT_MS);
 
         futures = new LinkedList<>();
         ser = new Serializer();
@@ -180,11 +182,6 @@ public class PoolingManagerImplTest {
         mgr = new PoolingManagerImpl(controller, poolProps);
     }
 
-    @After
-    public void tearDown() throws Exception {
-
-    }
-
     @Test
     public void testPoolingManagerImpl() {
         mgr = new PoolingManagerImpl(controller, poolProps);
@@ -199,8 +196,8 @@ public class PoolingManagerImplTest {
     @Test
     public void testPoolingManagerImpl_ClassEx() {
         /*
-         * this controller does not implement TopicListener, which should cause
-         * a ClassCastException
+         * this controller does not implement TopicListener, which should cause a
+         * ClassCastException
          */
         PolicyController ctlr = mock(PolicyController.class);
 
@@ -316,6 +313,7 @@ public class PoolingManagerImplTest {
 
         verify(dmaap).stopConsumer(mgr);
         verify(sched).shutdownNow();
+        verify(dmaap).publish(contains("offline"));
 
         assertTrue(mgr.getCurrent() instanceof IdleState);
     }
@@ -362,7 +360,7 @@ public class PoolingManagerImplTest {
         mgr.afterStop();
 
         verify(eventQueue).clear();
-        verify(dmaap).stopPublisher();
+        verify(dmaap).stopPublisher(STD_OFFLINE_PUB_WAIT_MS);
     }
 
     @Test
@@ -376,7 +374,7 @@ public class PoolingManagerImplTest {
         mgr.afterStop();
 
         verify(eventQueue, never()).clear();
-        verify(dmaap).stopPublisher();
+        verify(dmaap).stopPublisher(STD_OFFLINE_PUB_WAIT_MS);
     }
 
     @Test
@@ -511,7 +509,7 @@ public class PoolingManagerImplTest {
 
         CountDownLatch latch = new CountDownLatch(1);
 
-        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, xxx -> {
+        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, () -> {
             latch.countDown();
             return null;
         });
@@ -539,7 +537,7 @@ public class PoolingManagerImplTest {
 
         CountDownLatch latch = new CountDownLatch(1);
 
-        mgr.scheduleWithFixedDelay(STD_HEARTBEAT_WAIT_MS, STD_ACTIVE_HEARTBEAT_MS, xxx -> {
+        mgr.scheduleWithFixedDelay(STD_HEARTBEAT_WAIT_MS, STD_ACTIVE_HEARTBEAT_MS, () -> {
             latch.countDown();
             return null;
         });
@@ -608,8 +606,7 @@ public class PoolingManagerImplTest {
         StartState st = (StartState) mgr.getCurrent();
 
         /*
-         * give it its heart beat, that should cause it to transition to the
-         * Query state.
+         * give it its heart beat, that should cause it to transition to the Query state.
          */
         Heartbeat hb = new Heartbeat(mgr.getHost(), st.getHbTimestampMs());
         hb.setChannel(Message.ADMIN);
@@ -985,14 +982,35 @@ public class PoolingManagerImplTest {
     }
 
     @Test
+    public void testInject_Ex() throws Exception {
+        startMgr();
+
+        // 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());
+
+        CountDownLatch latch = catchRecursion(true);
+
+        Forward msg = new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID);
+        mgr.handle(msg);
+
+        verify(dmaap, times(START_PUB)).publish(any());
+        verify(controller).onTopicEvent(CommInfrastructure.UEB, TOPIC2, THE_EVENT);
+
+        // ensure we made it past both beforeXxx() methods
+        assertEquals(0, latch.getCount());
+    }
+
+    @Test
     public void testHandleInternal() throws Exception {
         startMgr();
 
         StartState st = (StartState) mgr.getCurrent();
 
         /*
-         * give it its heart beat, that should cause it to transition to the
-         * Query state.
+         * give it its heart beat, that should cause it to transition to the Query state.
          */
         Heartbeat hb = new Heartbeat(mgr.getHost(), st.getHbTimestampMs());
         hb.setChannel(Message.ADMIN);
@@ -1022,8 +1040,8 @@ public class PoolingManagerImplTest {
         Heartbeat hb = new Heartbeat(mgr.getHost(), st.getHbTimestampMs());
 
         /*
-         * do NOT set the channel - this will cause the message to be invalid,
-         * triggering an exception
+         * do NOT set the channel - this will cause the message to be invalid, triggering
+         * an exception
          */
 
         String msg = ser.encodeMsg(hb);
@@ -1068,7 +1086,7 @@ public class PoolingManagerImplTest {
         when(eventQueue.poll()).thenAnswer(args -> lst.poll());
 
         // route the messages to this host
-        mgr.startDistributing(makeAssignments(true));
+        assertTrue(mgr.startDistributing(makeAssignments(true)).await(2, TimeUnit.SECONDS));
 
         // all of the events should have been processed locally
         verify(dmaap, times(START_PUB)).publish(any());
@@ -1088,7 +1106,7 @@ public class PoolingManagerImplTest {
         when(eventQueue.poll()).thenAnswer(args -> lst.poll());
 
         // route the messages to the OTHER host
-        mgr.startDistributing(makeAssignments(false));
+        assertTrue(mgr.startDistributing(makeAssignments(false)).await(2, TimeUnit.SECONDS));
 
         // all of the events should have been forwarded
         verify(dmaap, times(4)).publish(any());
@@ -1140,7 +1158,7 @@ public class PoolingManagerImplTest {
 
         CountDownLatch latch = new CountDownLatch(1);
 
-        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, xxx -> {
+        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, () -> {
             latch.countDown();
             return null;
         });
@@ -1163,7 +1181,7 @@ public class PoolingManagerImplTest {
 
         CountDownLatch latch = new CountDownLatch(1);
 
-        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, xxx -> {
+        mgr.schedule(STD_ACTIVE_HEARTBEAT_MS, () -> {
             latch.countDown();
             return null;
         });
@@ -1208,15 +1226,14 @@ public class PoolingManagerImplTest {
     }
 
     /**
-     * Configure the mock controller to act like a real controller, invoking
-     * beforeOffer and then beforeInsert, so we can make sure they pass through.
-     * We'll keep count to ensure we don't get into infinite recursion.
+     * Configure the mock controller to act like a real controller, invoking beforeOffer
+     * and then beforeInsert, so we can make sure they pass through. We'll keep count to
+     * ensure we don't get into infinite recursion.
      * 
-     * @param invokeBeforeInsert {@code true} if beforeInsert() should be
-     *        invoked, {@code false} if it should be skipped
+     * @param invokeBeforeInsert {@code true} if beforeInsert() should be invoked,
+     *        {@code false} if it should be skipped
      * 
-     * @return a latch that will be counted down if both beforeXxx() methods
-     *         return false
+     * @return a latch that will be counted down if both beforeXxx() methods return false
      */
     private CountDownLatch catchRecursion(boolean invokeBeforeInsert) {
         CountDownLatch recursion = new CountDownLatch(3);
@@ -1230,9 +1247,9 @@ public class PoolingManagerImplTest {
             }
 
             int iarg = 0;
-            CommInfrastructure proto = args.getArgumentAt(iarg++, CommInfrastructure.class);
-            String topic = args.getArgumentAt(iarg++, String.class);
-            String event = args.getArgumentAt(iarg++, String.class);
+            CommInfrastructure proto = args.getArgument(iarg++);
+            String topic = args.getArgument(iarg++);
+            String event = args.getArgument(iarg++);
 
             if (mgr.beforeOffer(proto, topic, event)) {
                 return null;
@@ -1253,9 +1270,8 @@ public class PoolingManagerImplTest {
     /**
      * Makes an assignment with two buckets.
      * 
-     * @param sameHost {@code true} if the {@link #REQUEST_ID} should has to the
-     *        manager's bucket, {@code false} if it should hash to the other
-     *        host's bucket
+     * @param sameHost {@code true} if the {@link #REQUEST_ID} should hash to the
+     *        manager's bucket, {@code false} if it should hash to the other host's bucket
      * @return a new bucket assignment
      */
     private BucketAssignments makeAssignments(boolean sameHost) {
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 63eb59d4..2d734c1c 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
@@ -30,6 +30,7 @@ import static org.onap.policy.drools.pooling.PoolingProperties.IDENTIFICATION_MS
 import static org.onap.policy.drools.pooling.PoolingProperties.INTER_HEARTBEAT_MS;
 import static org.onap.policy.drools.pooling.PoolingProperties.OFFLINE_AGE_MS;
 import static org.onap.policy.drools.pooling.PoolingProperties.OFFLINE_LIMIT;
+import static org.onap.policy.drools.pooling.PoolingProperties.OFFLINE_PUB_WAIT_MS;
 import static org.onap.policy.drools.pooling.PoolingProperties.POOLING_TOPIC;
 import static org.onap.policy.drools.pooling.PoolingProperties.REACTIVATE_MS;
 import static org.onap.policy.drools.pooling.PoolingProperties.START_HEARTBEAT_MS;
@@ -53,6 +54,7 @@ public class PoolingPropertiesTest {
     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;
 
     private Properties plain;
     private PoolingProperties pooling;
@@ -121,10 +123,15 @@ 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 the property name can be generalized.
+     * 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
+     * the property name can be generalized.
      * 
      * @param propnm name of the property of interest
      * @param specValue expected specialized value
@@ -140,8 +147,8 @@ public class PoolingPropertiesTest {
         assertEquals("special " + propnm, specValue, func.apply(null));
 
         /*
-         * Ensure the property supports generalization - this will throw an
-         * exception if it does not.
+         * Ensure the property supports generalization - this will throw an exception if
+         * it does not.
          */
         assertFalse(propnm.equals(generalize(propnm)));
 
@@ -155,8 +162,8 @@ public class PoolingPropertiesTest {
     }
 
     /**
-     * Makes a set of properties, where all of the properties are specialized
-     * for the controller.
+     * Makes a set of properties, where all of the properties are specialized for the
+     * controller.
      * 
      * @return a new property set
      */
@@ -172,6 +179,7 @@ public class PoolingPropertiesTest {
         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 8b495099..505dc400 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
@@ -49,8 +49,8 @@ public class SpecPropertiesTest {
     private static final String PREFIX_SPEC = PREFIX_GEN + MY_SPEC + ".";
 
     /**
-     * Suffix to add to property names to generate names of properties that are
-     * not populated.
+     * Suffix to add to property names to generate names of properties that are not
+     * populated.
      */
     private static final String SUFFIX = ".suffix";
 
@@ -175,6 +175,16 @@ public class SpecPropertiesTest {
         assertNull(props.getProperty(gen(PROP_UNKNOWN), null));
     }
 
+    @Test(expected = UnsupportedOperationException.class)
+    public void testHashCode() {
+        props.hashCode();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testEquals() {
+        props.equals(props);
+    }
+
     private String gen(String propnm) {
         if (propnm.startsWith(PREFIX_SPEC)) {
             return PREFIX_GEN + propnm.substring(PREFIX_SPEC.length());
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/BucketAssignmentsTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/BucketAssignmentsTest.java
index ef03d4d6..c14e8dba 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/BucketAssignmentsTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/BucketAssignmentsTest.java
@@ -26,6 +26,7 @@ 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 java.util.Arrays;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import org.junit.Test;
@@ -196,16 +197,21 @@ public class BucketAssignmentsTest {
         String[] arr = {"def", "abc", "def", "ghi", "def", "def", "xyz"};
         asgn.setHostArray(arr);
 
-        for (int x = 0; x < arr.length; ++x) {
-            assertEquals("x=" + x, arr[x], asgn.getAssignedHost(x));
+        /*
+         * get assignments for consecutive integers, including negative numbers and
+         * numbers extending past the length of the array.
+         * 
+         */
+        TreeSet<String> seen = new TreeSet<>();
+        for (int x = -1; x < arr.length + 2; ++x) {
+            seen.add(asgn.getAssignedHost(x));
         }
 
-        // negative
-        assertNull(asgn.getAssignedHost(-1));
+        TreeSet<String> expected = new TreeSet<>(Arrays.asList(arr));
+        assertEquals(expected, seen);
 
-        // beyond end
-        assertNull(asgn.getAssignedHost(arr.length));
-        assertNull(asgn.getAssignedHost(arr.length + 1));
+        // try a much bigger number
+        assertNotNull(asgn.getAssignedHost(arr.length * 1000));
     }
 
     @Test
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/Trial.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/Trial.java
deleted file mode 100644
index 428b5853..00000000
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/message/Trial.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ================================================================================
- * 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.message;
-
-import org.junit.Test;
-import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-public class Trial {
-
-	@Test
-	public void test() throws Exception {
-		ObjectMapper mapper = new ObjectMapper();
-		
-		Message msg = new Forward("me", CommInfrastructure.DMAAP, "my topic", "a message", "my req");
-		
-		String enc = mapper.writeValueAsString(msg);
-		System.out.println("enc=" + enc);
-		
-		Message msg2 = mapper.readValue(enc, Message.class);
-		System.out.println("class=" + msg2.getClass());
-	}
-
-}
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 7997a4ee..7b4b0602 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
@@ -24,9 +24,9 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyLong;
-import static org.mockito.Matchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -39,11 +39,12 @@ import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.drools.pooling.message.BucketAssignments;
 import org.onap.policy.drools.pooling.message.Heartbeat;
-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.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 ActiveStateTest extends BasicStateTester {
 
@@ -65,7 +66,7 @@ public class ActiveStateTest extends BasicStateTester {
 
         // ensure a heart beat was generated
         Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
-        assertEquals(MY_HOST, msg.second.getSource());
+        assertEquals(MY_HOST, msg.second().getSource());
     }
 
     @Test
@@ -183,8 +184,8 @@ public class ActiveStateTest extends BasicStateTester {
 
         /*
          * 
-         * PREV_HOST2 has buckets and is my predecessor, but it isn't the leader
-         * thus should be ignored.
+         * PREV_HOST2 has buckets and is my predecessor, but it isn't the leader thus
+         * should be ignored.
          */
         assertNull(state.process(new Offline(PREV_HOST2)));
     }
@@ -196,22 +197,57 @@ public class ActiveStateTest extends BasicStateTester {
         state = new ActiveState(mgr);
 
         /*
-         * HOST1 has buckets, but it isn't the leader and it isn't my
-         * predecessor, thus should be ignored.
+         * HOST1 has buckets, but it isn't the leader and it isn't my predecessor, thus
+         * should be ignored.
          */
         assertNull(state.process(new Offline(HOST1)));
     }
 
     @Test
-    public void testProcessQuery() {
+    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();
+
+        // info should be unchanged
+        assertEquals(MY_HOST, state.getLeader());
+        assertEquals(ASGN3, state.getAssignments());
+    }
+
+    @Test
+    public void testProcessLeader_BadLeader() {
+        String[] arr = {HOST2, HOST1};
+        BucketAssignments asgn = new BucketAssignments(arr);
+
+        // now send a Leader message for that leader
+        Leader msg = new Leader(HOST1, asgn);
+
         State next = mock(State.class);
         when(mgr.goQuery()).thenReturn(next);
 
-        assertEquals(next, state.process(new Query()));
+        // should go Query, but not start distributing
+        assertEquals(next, state.process(msg));
+        verify(mgr, never()).startDistributing(asgn);
+    }
+
+    @Test
+    public void testProcessLeader_GoodLeader() {
+        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
+
+        // now send a Leader message for that leader
+        Leader msg = new Leader(PREV_HOST, asgn);
+
+        State next = mock(State.class);
+        when(mgr.goActive()).thenReturn(next);
 
-        Identification ident = captureAdminMessage(Identification.class);
-        assertEquals(MY_HOST, ident.getSource());
-        assertEquals(ASGN3, ident.getAssignments());
+        // should go Active and start distributing
+        assertEquals(next, state.process(msg));
+        verify(mgr).startDistributing(asgn);
     }
 
     @Test
@@ -256,24 +292,24 @@ public class ActiveStateTest extends BasicStateTester {
         // invoke start() to add the timers
         state.start();
 
-        assertEquals(3, repeatedFutures.size());
+        assertEquals(3, repeatedSchedules.size());
 
         Triple<Long, Long, StateTimerTask> timer;
 
         // heart beat generator
         timer = repeatedTasks.remove();
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first.longValue());
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second.longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_ACTIVE_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_INTER_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_INTER_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_INTER_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
     }
 
     @Test
@@ -285,19 +321,19 @@ public class ActiveStateTest extends BasicStateTester {
         // invoke start() to add the timers
         state.start();
 
-        assertEquals(2, repeatedFutures.size());
+        assertEquals(2, repeatedSchedules.size());
 
         Triple<Long, Long, StateTimerTask> timer;
 
         // heart beat generator
         timer = repeatedTasks.remove();
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first.longValue());
-        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.second.longValue());
+        assertEquals(STD_ACTIVE_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_ACTIVE_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_INTER_HEARTBEAT_MS, timer.first().longValue());
+        assertEquals(STD_INTER_HEARTBEAT_MS, timer.second().longValue());
     }
 
     @Test
@@ -314,14 +350,14 @@ public class ActiveStateTest extends BasicStateTester {
         verify(mgr).publish(anyString(), any(Heartbeat.class));
 
         // fire the task
-        assertNull(task.third.fire(null));
+        assertNull(task.third().fire());
 
         // should have generated a second pair of heart beats
         verify(mgr, times(2)).publish(anyString(), any(Heartbeat.class));
 
         Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
-        assertEquals(MY_HOST, msg.first);
-        assertEquals(MY_HOST, msg.second.getSource());
+        assertEquals(MY_HOST, msg.first());
+        assertEquals(MY_HOST, msg.second().getSource());
     }
 
     @Test
@@ -339,7 +375,7 @@ public class ActiveStateTest extends BasicStateTester {
         when(mgr.goInactive()).thenReturn(next);
 
         // fire the task - should not transition
-        assertNull(task.third.fire(null));
+        assertNull(task.third().fire());
 
         verify(mgr, never()).publishAdmin(any(Query.class));
     }
@@ -356,7 +392,7 @@ public class ActiveStateTest extends BasicStateTester {
         when(mgr.goInactive()).thenReturn(next);
 
         // fire the task - should transition
-        assertEquals(next, task.third.fire(null));
+        assertEquals(next, task.third().fire());
 
         // should indicate failure
         verify(mgr).internalTopicFailed();
@@ -381,7 +417,7 @@ public class ActiveStateTest extends BasicStateTester {
         when(mgr.goQuery()).thenReturn(next);
 
         // fire the task - should NOT transition
-        assertNull(task.third.fire(null));
+        assertNull(task.third().fire());
 
         verify(mgr, never()).publishAdmin(any(Query.class));
     }
@@ -398,7 +434,7 @@ public class ActiveStateTest extends BasicStateTester {
         when(mgr.goQuery()).thenReturn(next);
 
         // fire the task - should transition
-        assertEquals(next, task.third.fire(null));
+        assertEquals(next, task.third().fire());
 
         verify(mgr).publishAdmin(any(Query.class));
     }
@@ -414,8 +450,8 @@ public class ActiveStateTest extends BasicStateTester {
         verify(mgr, times(1)).publish(any(), any());
 
         Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
-        assertEquals(MY_HOST, msg.first);
-        assertEquals(MY_HOST, msg.second.getSource());
+        assertEquals(MY_HOST, msg.first());
+        assertEquals(MY_HOST, msg.second().getSource());
     }
 
     @Test
@@ -429,13 +465,13 @@ public class ActiveStateTest extends BasicStateTester {
 
         // this message should go to itself
         msg = capturePublishedMessage(Heartbeat.class, index++);
-        assertEquals(MY_HOST, msg.first);
-        assertEquals(MY_HOST, msg.second.getSource());
+        assertEquals(MY_HOST, msg.first());
+        assertEquals(MY_HOST, msg.second().getSource());
 
         // this message should go to its successor
         msg = capturePublishedMessage(Heartbeat.class, index++);
-        assertEquals(HOST1, msg.first);
-        assertEquals(MY_HOST, msg.second.getSource());
+        assertEquals(HOST1, msg.first());
+        assertEquals(MY_HOST, msg.second().getSource());
     }
 
 }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/BasicStateTester.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/BasicStateTester.java
index e48742f7..75ca7564 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/BasicStateTester.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/BasicStateTester.java
@@ -21,9 +21,9 @@
 package org.onap.policy.drools.pooling.state;
 
 import static org.junit.Assert.assertNotNull;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyLong;
-import static org.mockito.Matchers.anyString;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -33,16 +33,18 @@ import java.util.List;
 import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
-import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicReference;
+import org.onap.policy.drools.pooling.CancellableScheduledTask;
 import org.onap.policy.drools.pooling.PoolingManager;
 import org.onap.policy.drools.pooling.PoolingProperties;
 import org.onap.policy.drools.pooling.message.BucketAssignments;
 import org.onap.policy.drools.pooling.message.Leader;
 import org.onap.policy.drools.pooling.message.Message;
+import org.onap.policy.drools.utils.Pair;
+import org.onap.policy.drools.utils.Triple;
 
 /**
- * Superclass used to test subclasses of {@link Message}.
+ * Superclass used to test subclasses of {@link State}.
  */
 public class BasicStateTester {
 
@@ -74,9 +76,9 @@ public class BasicStateTester {
     protected static final BucketAssignments ASGN3 = new BucketAssignments(HOST_ARR3);
 
     /**
-     * Futures returned by schedule().
+     * Scheduled tasks returned by schedule().
      */
-    protected LinkedList<ScheduledFuture<?>> onceFutures;
+    protected LinkedList<CancellableScheduledTask> onceSchedules;
 
     /**
      * Tasks captured via schedule().
@@ -84,9 +86,9 @@ public class BasicStateTester {
     protected LinkedList<Pair<Long, StateTimerTask>> onceTasks;
 
     /**
-     * Futures returned by scheduleWithFixedDelay().
+     * Scheduled tasks returned by scheduleWithFixedDelay().
      */
-    protected LinkedList<ScheduledFuture<?>> repeatedFutures;
+    protected LinkedList<CancellableScheduledTask> repeatedSchedules;
 
     /**
      * Tasks captured via scheduleWithFixedDelay().
@@ -112,10 +114,10 @@ public class BasicStateTester {
     }
 
     public void setUp() throws Exception {
-        onceFutures = new LinkedList<>();
+        onceSchedules = new LinkedList<>();
         onceTasks = new LinkedList<>();
 
-        repeatedFutures = new LinkedList<>();
+        repeatedSchedules = new LinkedList<>();
         repeatedTasks = new LinkedList<>();
 
         published = new LinkedList<>();
@@ -162,9 +164,9 @@ public class BasicStateTester {
             Object[] args = invocation.getArguments();
             onceTasks.add(new Pair<>((Long) args[0], (StateTimerTask) args[1]));
 
-            ScheduledFuture<?> fut = mock(ScheduledFuture.class);
-            onceFutures.add(fut);
-            return fut;
+            CancellableScheduledTask sched = mock(CancellableScheduledTask.class);
+            onceSchedules.add(sched);
+            return sched;
         });
 
         // capture scheduleWithFixedDelay() arguments, and return a new future
@@ -172,9 +174,9 @@ public class BasicStateTester {
             Object[] args = invocation.getArguments();
             repeatedTasks.add(new Triple<>((Long) args[0], (Long) args[1], (StateTimerTask) args[2]));
 
-            ScheduledFuture<?> fut = mock(ScheduledFuture.class);
-            repeatedFutures.add(fut);
-            return fut;
+            CancellableScheduledTask sched = mock(CancellableScheduledTask.class);
+            repeatedSchedules.add(sched);
+            return sched;
         });
 
         // get/set assignments in the manager
@@ -183,7 +185,7 @@ public class BasicStateTester {
         when(mgr.getAssignments()).thenAnswer(args -> asgn.get());
 
         doAnswer(args -> {
-            asgn.set(args.getArgumentAt(0, BucketAssignments.class));
+            asgn.set(args.getArgument(0));
             return null;
         }).when(mgr).startDistributing(any());
     }
@@ -199,8 +201,7 @@ public class BasicStateTester {
     }
 
     /**
-     * Captures the host array from the Leader message published to the admin
-     * channel.
+     * Captures the host array from the Leader message published to the admin channel.
      * 
      * @return the host array, as a list
      */
@@ -209,8 +210,7 @@ public class BasicStateTester {
     }
 
     /**
-     * Captures the host array from the Leader message published to the admin
-     * channel.
+     * Captures the host array from the Leader message published to the admin channel.
      * 
      * @return the host array
      */
@@ -224,8 +224,7 @@ public class BasicStateTester {
     }
 
     /**
-     * Captures the assignments from the Leader message published to the admin
-     * channel.
+     * Captures the assignments from the Leader message published to the admin channel.
      * 
      * @return the bucket assignments
      */
@@ -277,42 +276,6 @@ public class BasicStateTester {
      */
     protected <T extends Message> Pair<String, T> capturePublishedMessage(Class<T> clazz, int index) {
         Pair<String, Message> msg = published.get(index);
-        return new Pair<>(msg.first, clazz.cast(msg.second));
+        return new Pair<>(msg.first(), clazz.cast(msg.second()));
     }
-
-    /**
-     * Pair of values.
-     * 
-     * @param <F> first value's type
-     * @param <S> second value's type
-     */
-    public static class Pair<F, S> {
-        public final F first;
-        public final S second;
-
-        public Pair(F first, S second) {
-            this.first = first;
-            this.second = second;
-        }
-    }
-
-    /**
-     * Pair of values.
-     * 
-     * @param <F> first value's type
-     * @param <S> second value's type
-     * @param <T> third value's type
-     */
-    public static class Triple<F, S, T> {
-        public final F first;
-        public final S second;
-        public final T third;
-
-        public Triple(F first, S second, T third) {
-            this.first = first;
-            this.second = second;
-            this.third = third;
-        }
-    }
-
 }
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/IdleStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/IdleStateTest.java
index 96c59719..95cbe753 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/IdleStateTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/IdleStateTest.java
@@ -61,12 +61,6 @@ public class IdleStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testStop() {
-        state.stop();
-        verifyNothingPublished();
-    }
-
-    @Test
     public void testProcessForward() {
         Forward msg = new Forward();
         assertNull(state.process(msg));
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 48d5b1ed..394adaee 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
@@ -28,6 +28,7 @@ import java.util.Map;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.drools.pooling.message.Message;
+import org.onap.policy.drools.utils.Pair;
 
 public class InactiveStateTest extends BasicStateTester {
 
@@ -62,20 +63,20 @@ public class InactiveStateTest extends BasicStateTester {
 
         Pair<Long, StateTimerTask> timer = onceTasks.remove();
 
-        assertEquals(STD_REACTIVATE_WAIT_MS, timer.first.longValue());
+        assertEquals(STD_REACTIVATE_WAIT_MS, timer.first().longValue());
 
         // invoke the task - it should go to the state returned by the mgr
         State next = mock(State.class);
         when(mgr.goStart()).thenReturn(next);
 
-        assertEquals(next, timer.second.fire(null));
+        assertEquals(next, timer.second().fire());
     }
 
     @Test
     public void testInactiveState() {
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
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 d60ad2ea..e1718418 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
@@ -24,7 +24,7 @@ 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.Matchers.any;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -38,16 +38,19 @@ 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.pooling.state.ProcessingState.HostBucket;
 
 public class ProcessingStateTest extends BasicStateTester {
 
     private ProcessingState state;
+    private HostBucket hostBucket;
 
     @Before
     public void setUp() throws Exception {
         super.setUp();
 
         state = new ProcessingState(mgr, MY_HOST);
+        hostBucket = new HostBucket(MY_HOST);
     }
 
     @Test
@@ -62,6 +65,39 @@ 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);
@@ -97,8 +133,8 @@ public class ProcessingStateTest extends BasicStateTester {
         state = new ProcessingState(mgr, LEADER);
 
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
 
@@ -260,8 +296,8 @@ public class ProcessingStateTest extends BasicStateTester {
     @Test
     public void testMakeBucketArray() {
         /*
-         * All hosts are still alive, so it should have the exact same
-         * assignments as it had to start.
+         * All hosts are still alive, so it should have the exact same assignments as it
+         * had to start.
          */
         state.setAssignments(ASGN3);
         state.becomeLeader(sortHosts(HOST_ARR3));
@@ -325,4 +361,80 @@ public class ProcessingStateTest extends BasicStateTester {
         assertEquals(Arrays.asList(expected), captureHostList());
     }
 
+    @Test
+    public void testHostBucketRemove_testHostBucketAdd_testHostBucketSize() {
+        assertEquals(0, hostBucket.size());
+
+        hostBucket.add(20);
+        hostBucket.add(30);
+        hostBucket.add(40);
+        assertEquals(3, hostBucket.size());
+
+        assertEquals(20, hostBucket.remove().intValue());
+        assertEquals(30, hostBucket.remove().intValue());
+        assertEquals(1, hostBucket.size());
+
+        // add more before taking the last item
+        hostBucket.add(50);
+        hostBucket.add(60);
+        assertEquals(3, hostBucket.size());
+
+        assertEquals(40, hostBucket.remove().intValue());
+        assertEquals(50, hostBucket.remove().intValue());
+        assertEquals(60, hostBucket.remove().intValue());
+        assertEquals(0, hostBucket.size());
+
+        // add more AFTER taking the last item
+        hostBucket.add(70);
+        assertEquals(70, hostBucket.remove().intValue());
+        assertEquals(0, hostBucket.size());
+    }
+
+    @Test
+    public void testHostBucketCompareTo() {
+        HostBucket hb1 = new HostBucket(PREV_HOST);
+        HostBucket hb2 = new HostBucket(MY_HOST);
+
+        assertEquals(0, hb1.compareTo(hb1));
+        assertEquals(0, hb1.compareTo(new HostBucket(PREV_HOST)));
+
+        // both empty
+        assertTrue(hb1.compareTo(hb2) < 0);
+        assertTrue(hb2.compareTo(hb1) > 0);
+
+        // hb1 has one bucket, so it should not be larger
+        hb1.add(100);
+        assertTrue(hb1.compareTo(hb2) > 0);
+        assertTrue(hb2.compareTo(hb1) < 0);
+
+        // hb1 has two buckets, so it should still be larger
+        hb1.add(200);
+        assertTrue(hb1.compareTo(hb2) > 0);
+        assertTrue(hb2.compareTo(hb1) < 0);
+
+        // hb1 has two buckets, hb2 has one, so hb1 should still be larger
+        hb2.add(1000);
+        assertTrue(hb1.compareTo(hb2) > 0);
+        assertTrue(hb2.compareTo(hb1) < 0);
+
+        // same number of buckets, so hb2 should now be larger
+        hb2.add(2000);
+        assertTrue(hb1.compareTo(hb2) < 0);
+        assertTrue(hb2.compareTo(hb1) > 0);
+
+        // hb2 has more buckets, it should still be larger
+        hb2.add(3000);
+        assertTrue(hb1.compareTo(hb2) < 0);
+        assertTrue(hb2.compareTo(hb1) > 0);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testHostBucketHashCode() {
+        hostBucket.hashCode();
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void testHostBucketEquals() {
+        hostBucket.equals(hostBucket);
+    }
 }
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 d714d5cc..a7c3a3d5 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
@@ -25,9 +25,10 @@ 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.mockito.Matchers.any;
+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;
@@ -38,7 +39,7 @@ 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.Offline;
-import org.onap.policy.drools.pooling.message.Query;
+import org.onap.policy.drools.utils.Pair;
 
 public class QueryStateTest extends BasicStateTester {
 
@@ -63,44 +64,48 @@ public class QueryStateTest extends BasicStateTester {
     }
 
     @Test
+    public void testGoQuery() {
+        assertNull(state.goQuery());
+    }
+
+    @Test
     public void testStart() {
         state.start();
 
         Pair<Long, StateTimerTask> timer = onceTasks.remove();
 
-        assertEquals(STD_IDENTIFICATION_MS, timer.first.longValue());
-        assertNotNull(timer.second);
+        assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
+        assertNotNull(timer.second());
     }
 
     @Test
-    public void testGoQuery() {
-        assertNull(state.process(new Query()));
-        assertEquals(ASGN3, state.getAssignments());
-    }
+    public void testProcessIdentification_SameSource() {
+        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
 
-    @Test
-    public void testProcessIdentification_NullSource() {
-        assertNull(state.process(new Identification()));
+        assertNull(state.process(new Identification(MY_HOST, asgn)));
 
+        // info should be unchanged
         assertEquals(MY_HOST, state.getLeader());
+        verify(mgr, never()).startDistributing(asgn);
     }
 
     @Test
-    public void testProcessIdentification_NewLeader() {
-        assertNull(state.process(new Identification(PREV_HOST, null)));
-
-        assertEquals(PREV_HOST, state.getLeader());
-    }
+    public void testProcessIdentification_DiffSource() {
+        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
 
-    @Test
-    public void testProcessIdentification_NotNewLeader() {
-        assertNull(state.process(new Identification(HOST2, null)));
+        assertNull(state.process(new Identification(HOST2, asgn)));
 
+        // leader should be unchanged
         assertEquals(MY_HOST, state.getLeader());
+
+        // should have picked up the assignments
+        verify(mgr).startDistributing(asgn);
     }
 
     @Test
-    public void testProcessLeader_NullAssignment() {
+    public void testProcessLeader_Invalid() {
         Leader msg = new Leader(PREV_HOST, null);
 
         // should stay in the same state, and not start distributing
@@ -115,67 +120,55 @@ public class QueryStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testProcessLeader_NullSource() {
+    public void testProcessLeader_SameLeader() {
         String[] arr = {HOST2, PREV_HOST, MY_HOST};
         BucketAssignments asgn = new BucketAssignments(arr);
-        Leader msg = new Leader(null, asgn);
 
-        // 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();
+        // identify a leader that's better than my host
+        assertEquals(null, state.process(new Identification(PREV_HOST, asgn)));
 
-        // info should be unchanged
-        assertEquals(MY_HOST, state.getLeader());
-        assertEquals(ASGN3, state.getAssignments());
-    }
+        // now send a Leader message for that leader
+        Leader msg = new Leader(PREV_HOST, asgn);
 
-    @Test
-    public void testProcessLeader_SourceIsNotAssignmentLeader() {
-        String[] arr = {HOST2, PREV_HOST, MY_HOST};
-        BucketAssignments asgn = new BucketAssignments(arr);
-        Leader msg = new Leader(HOST2, asgn);
+        State next = mock(State.class);
+        when(mgr.goActive()).thenReturn(next);
 
-        // should stay in the same state, and not start distributing
-        assertNull(state.process(msg));
-        verify(mgr, never()).startDistributing(any());
-        verify(mgr, never()).goActive();
+        // should go Active and start distributing
+        assertEquals(next, state.process(msg));
         verify(mgr, never()).goInactive();
 
-        // info should be unchanged
-        assertEquals(MY_HOST, state.getLeader());
-        assertEquals(ASGN3, state.getAssignments());
+        // Ident msg + Leader msg = times(2)
+        verify(mgr, times(2)).startDistributing(asgn);
     }
 
     @Test
-    public void testProcessLeader_EmptyAssignment() {
-        Leader msg = new Leader(PREV_HOST, new BucketAssignments());
+    public void testProcessLeader_BetterLeaderWithAssignment() {
+        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
+        Leader msg = new Leader(PREV_HOST, asgn);
 
-        // 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();
+        State next = mock(State.class);
+        when(mgr.goActive()).thenReturn(next);
 
-        // info should be unchanged
-        assertEquals(MY_HOST, state.getLeader());
-        assertEquals(ASGN3, state.getAssignments());
+        // should go Active and start distributing
+        assertEquals(next, state.process(msg));
+        verify(mgr).startDistributing(asgn);
+        verify(mgr, never()).goInactive();
     }
 
     @Test
-    public void testProcessLeader_BetterLeader() {
-        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+    public void testProcessLeader_BetterLeaderWithoutAssignment() {
+        String[] arr = {HOST2, PREV_HOST, HOST1};
         BucketAssignments asgn = new BucketAssignments(arr);
         Leader msg = new Leader(PREV_HOST, asgn);
 
         State next = mock(State.class);
-        when(mgr.goActive()).thenReturn(next);
+        when(mgr.goInactive()).thenReturn(next);
 
-        // should go Active and start distributing
+        // should go Inactive, but start distributing
         assertEquals(next, state.process(msg));
         verify(mgr).startDistributing(asgn);
-        verify(mgr, never()).goInactive();
+        verify(mgr, never()).goActive();
     }
 
     @Test
@@ -241,41 +234,48 @@ public class QueryStateTest extends BasicStateTester {
     }
 
     @Test
-    public void testProcessQuery() {
-        BucketAssignments asgn = new BucketAssignments(new String[] {HOST1, HOST2});
-        mgr.startDistributing(asgn);
-        state = new QueryState(mgr);
-
-        State next = mock(State.class);
-        when(mgr.goQuery()).thenReturn(next);
-
-        assertEquals(null, state.process(new Query()));
-
-        verify(mgr).publishAdmin(any(Identification.class));
-    }
-
-    @Test
     public void testQueryState() {
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
 
     @Test
+    public void testAwaitIdentification_MissingSelfIdent() {
+        state.start();
+
+        Pair<Long, StateTimerTask> timer = onceTasks.remove();
+
+        assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
+        assertNotNull(timer.second());
+
+        // should published an Offline message and go inactive
+
+        State next = mock(State.class);
+        when(mgr.goInactive()).thenReturn(next);
+
+        assertEquals(next, timer.second().fire());
+
+        Offline msg = captureAdminMessage(Offline.class);
+        assertEquals(MY_HOST, msg.getSource());
+    }
+
+    @Test
     public void testAwaitIdentification_Leader() {
         state.start();
+        state.process(new Identification(MY_HOST, null));
 
         Pair<Long, StateTimerTask> timer = onceTasks.remove();
 
-        assertEquals(STD_IDENTIFICATION_MS, timer.first.longValue());
-        assertNotNull(timer.second);
+        assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
+        assertNotNull(timer.second());
 
         State next = mock(State.class);
         when(mgr.goActive()).thenReturn(next);
 
-        assertEquals(next, timer.second.fire(null));
+        assertEquals(next, timer.second().fire());
 
         // should have published a Leader message
         Leader msg = captureAdminMessage(Leader.class);
@@ -291,20 +291,21 @@ public class QueryStateTest extends BasicStateTester {
         state = new QueryState(mgr);
 
         state.start();
+        state.process(new Identification(MY_HOST, null));
 
         // tell it the leader is still active
         state.process(new Identification(PREV_HOST, asgn));
 
         Pair<Long, StateTimerTask> timer = onceTasks.remove();
 
-        assertEquals(STD_IDENTIFICATION_MS, timer.first.longValue());
-        assertNotNull(timer.second);
+        assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
+        assertNotNull(timer.second());
 
         // set up active state, as that's what it should return
         State next = mock(State.class);
         when(mgr.goActive()).thenReturn(next);
 
-        assertEquals(next, timer.second.fire(null));
+        assertEquals(next, timer.second().fire());
 
         // should NOT have published a Leader message
         assertTrue(admin.isEmpty());
@@ -321,20 +322,21 @@ public class QueryStateTest extends BasicStateTester {
         state = new QueryState(mgr);
 
         state.start();
+        state.process(new Identification(MY_HOST, null));
 
         // tell it the leader is still active
         state.process(new Identification(PREV_HOST, asgn));
 
         Pair<Long, StateTimerTask> timer = onceTasks.remove();
 
-        assertEquals(STD_IDENTIFICATION_MS, timer.first.longValue());
-        assertNotNull(timer.second);
+        assertEquals(STD_IDENTIFICATION_MS, timer.first().longValue());
+        assertNotNull(timer.second());
 
         // set up inactive state, as that's what it should return
         State next = mock(State.class);
         when(mgr.goInactive()).thenReturn(next);
 
-        assertEquals(next, timer.second.fire(null));
+        assertEquals(next, timer.second().fire());
 
         // should NOT have published a Leader message
         assertTrue(admin.isEmpty());
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 f29d2348..af4e8f13 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
@@ -23,7 +23,7 @@ package org.onap.policy.drools.pooling.state;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -38,6 +38,7 @@ import org.onap.policy.drools.pooling.message.Leader;
 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;
 
 public class StartStateTest extends BasicStateTester {
 
@@ -74,18 +75,18 @@ public class StartStateTest extends BasicStateTester {
 
         Pair<String, Heartbeat> msg = capturePublishedMessage(Heartbeat.class);
 
-        assertEquals(MY_HOST, msg.first);
-        assertEquals(state.getHbTimestampMs(), msg.second.getTimestampMs());
+        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());
+        assertEquals(STD_HEARTBEAT_WAIT_MS, timer.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(null));
+        assertEquals(next, timer.second().fire());
 
         verify(mgr).internalTopicFailed();
     }
@@ -93,8 +94,8 @@ public class StartStateTest extends BasicStateTester {
     @Test
     public void testStartStatePoolingManager() {
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
@@ -105,8 +106,8 @@ public class StartStateTest extends BasicStateTester {
         state = new StartState(mgr);
 
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
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 1be48e21..a184dfad 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
@@ -21,16 +21,18 @@
 package org.onap.policy.drools.pooling.state;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
-import static org.mockito.Matchers.any;
+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.verify;
 import static org.mockito.Mockito.when;
 import java.util.Map;
-import java.util.concurrent.ScheduledFuture;
 import org.junit.Before;
 import org.junit.Test;
+import org.onap.policy.drools.pooling.CancellableScheduledTask;
 import org.onap.policy.drools.pooling.PoolingManager;
 import org.onap.policy.drools.pooling.message.BucketAssignments;
 import org.onap.policy.drools.pooling.message.Forward;
@@ -55,8 +57,8 @@ public class StateTest extends BasicStateTester {
     @Test
     public void testStatePoolingManager() {
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
@@ -67,8 +69,8 @@ public class StateTest extends BasicStateTester {
         state = new MyState(mgr);
 
         /*
-         * Prove the state is attached to the manager by invoking getHost(),
-         * which delegates to the manager.
+         * Prove the state is attached to the manager by invoking getHost(), which
+         * delegates to the manager.
          */
         assertEquals(MY_HOST, state.getHost());
     }
@@ -98,13 +100,13 @@ public class StateTest extends BasicStateTester {
         verify(mgr).schedule(delay, task2);
         verify(mgr).scheduleWithFixedDelay(initDelay, delay, task3);
 
-        ScheduledFuture<?> fut1 = onceFutures.removeFirst();
-        ScheduledFuture<?> fut2 = onceFutures.removeFirst();
-        ScheduledFuture<?> fut3 = repeatedFutures.removeFirst();
+        CancellableScheduledTask sched1 = onceSchedules.removeFirst();
+        CancellableScheduledTask sched2 = onceSchedules.removeFirst();
+        CancellableScheduledTask sched3 = repeatedSchedules.removeFirst();
 
-        verify(fut1, never()).cancel(false);
-        verify(fut2, never()).cancel(false);
-        verify(fut3, never()).cancel(false);
+        verify(sched1, never()).cancel();
+        verify(sched2, never()).cancel();
+        verify(sched3, never()).cancel();
 
         /*
          * Cancel the timers.
@@ -112,9 +114,9 @@ public class StateTest extends BasicStateTester {
         state.cancelTimers();
 
         // verify that all were cancelled
-        verify(fut1).cancel(false);
-        verify(fut2).cancel(false);
-        verify(fut3).cancel(false);
+        verify(sched1).cancel();
+        verify(sched2).cancel();
+        verify(sched3).cancel();
     }
 
     @Test
@@ -134,13 +136,6 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
-    public void testStop() {
-        state.stop();
-
-        assertEquals(MY_HOST, captureAdminMessage(Offline.class).getSource());
-    }
-
-    @Test
     public void testGoStart() {
         State next = mock(State.class);
         when(mgr.goStart()).thenReturn(next);
@@ -195,7 +190,18 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
-    public void testProcessLeader_NullAssignment() {
+    public void testProcessLeader() {
+        String[] arr = {HOST2, HOST1};
+        BucketAssignments asgn = new BucketAssignments(arr);
+        Leader msg = new Leader(HOST1, asgn);
+
+        // should ignore it
+        assertEquals(null, 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
@@ -206,57 +212,44 @@ public class StateTest extends BasicStateTester {
     }
 
     @Test
-    public void testProcessLeader_NullSource() {
+    public void testIsValidLeader_NullAssignment() {
+        assertFalse(state.isValid(new Leader(PREV_HOST, null)));
+    }
+
+    @Test
+    public void testIsValidLeader_NullSource() {
         String[] arr = {HOST2, PREV_HOST, MY_HOST};
         BucketAssignments asgn = new BucketAssignments(arr);
-        Leader msg = new Leader(null, asgn);
+        assertFalse(state.isValid(new Leader(null, asgn)));
+    }
 
-        // 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 testIsValidLeader_EmptyAssignment() {
+        assertFalse(state.isValid(new Leader(PREV_HOST, new BucketAssignments())));
     }
 
     @Test
-    public void testProcessLeader_EmptyAssignment() {
-        Leader msg = new Leader(PREV_HOST, new BucketAssignments());
+    public void testIsValidLeader_FromSelf() {
+        String[] arr = {HOST2, MY_HOST};
+        BucketAssignments asgn = new BucketAssignments(arr);
 
-        // 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();
+        assertFalse(state.isValid(new Leader(MY_HOST, asgn)));
     }
 
     @Test
-    public void testProcessLeader_MyHostAssigned() {
-        String[] arr = {HOST2, PREV_HOST, MY_HOST};
+    public void testIsValidLeader_WrongLeader() {
+        String[] arr = {HOST2, HOST3};
         BucketAssignments asgn = new BucketAssignments(arr);
-        Leader msg = new Leader(PREV_HOST, asgn);
-
-        State next = mock(State.class);
-        when(mgr.goActive()).thenReturn(next);
 
-        // should go Active and start distributing
-        assertEquals(next, state.process(msg));
-        verify(mgr).startDistributing(asgn);
-        verify(mgr, never()).goInactive();
+        assertFalse(state.isValid(new Leader(HOST1, asgn)));
     }
 
     @Test
-    public void testProcessLeader_MyHostUnassigned() {
+    public void testIsValidLeader() {
         String[] arr = {HOST2, HOST1};
         BucketAssignments asgn = new BucketAssignments(arr);
-        Leader msg = new Leader(HOST1, asgn);
 
-        State next = mock(State.class);
-        when(mgr.goInactive()).thenReturn(next);
-
-        // should go Inactive and start distributing
-        assertEquals(next, state.process(msg));
-        verify(mgr).startDistributing(asgn);
-        verify(mgr, never()).goActive();
+        assertTrue(state.isValid(new Leader(HOST1, asgn)));
     }
 
     @Test
@@ -344,18 +337,18 @@ public class StateTest extends BasicStateTester {
 
         state.schedule(delay, task);
 
-        ScheduledFuture<?> fut = onceFutures.removeFirst();
+        CancellableScheduledTask sched = onceSchedules.removeFirst();
 
         // scheduled, but not canceled yet
         verify(mgr).schedule(delay, task);
-        verify(fut, never()).cancel(false);
+        verify(sched, never()).cancel();
 
         /*
-         * Ensure the state added the timer to its list by telling it to cancel
-         * its timers and then seeing if this timer was canceled.
+         * Ensure the state added the timer to its list by telling it to cancel its timers
+         * and then seeing if this timer was canceled.
          */
         state.cancelTimers();
-        verify(fut).cancel(false);
+        verify(sched).cancel();
     }
 
     @Test
@@ -367,18 +360,18 @@ public class StateTest extends BasicStateTester {
 
         state.scheduleWithFixedDelay(initdel, delay, task);
 
-        ScheduledFuture<?> fut = repeatedFutures.removeFirst();
+        CancellableScheduledTask sched = repeatedSchedules.removeFirst();
 
         // scheduled, but not canceled yet
         verify(mgr).scheduleWithFixedDelay(initdel, delay, task);
-        verify(fut, never()).cancel(false);
+        verify(sched, never()).cancel();
 
         /*
-         * Ensure the state added the timer to its list by telling it to cancel
-         * its timers and then seeing if this timer was canceled.
+         * Ensure the state added the timer to its list by telling it to cancel its timers
+         * and then seeing if this timer was canceled.
          */
         state.cancelTimers();
-        verify(fut).cancel(false);
+        verify(sched).cancel();
     }
 
     @Test