Don't stop controller on dmaap failure

Modified feature-pooling-dmaap so that it continues to work even
if it's unable to communicate with the internal DMaaP topic.  When
that happens, it simply starts processing locally, until communication
is re-established and it receives a bucket assignment.
Fixed typo in comment.
Added to comment for State.internalTopicFailed().
Removed extra space before @Ignore.

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

7 files changed, 69 insertions, 351 deletions

diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/EventQueueTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/EventQueueTest.java
deleted file mode 100644
index 24144686..00000000
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/EventQueueTest.java
+++ /dev/null
@@ -1,196 +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.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import java.util.LinkedList;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.drools.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.drools.pooling.message.Forward;
-
-public class EventQueueTest {
-
-    private static final int MAX_SIZE = 5;
-    private static final long MAX_AGE_MS = 3000L;
-
-    private static final String MY_SOURCE = "my.source";
-    private static final CommInfrastructure MY_PROTO = CommInfrastructure.UEB;
-    private static final String MY_TOPIC = "my.topic";
-    private static final String MY_PAYLOAD = "my.payload";
-    private static final String MY_REQID = "my.request.id";
-
-    private EventQueue queue;
-
-    @Before
-    public void setUp() {
-        queue = new EventQueue(MAX_SIZE, MAX_AGE_MS);
-
-    }
-
-    @Test
-    public void testEventQueue() {
-        // shouldn't generate an exception
-        new EventQueue(1, 1);
-    }
-
-    @Test
-    public void testClear() {
-        // add some items
-        queue.add(makeActive());
-        queue.add(makeActive());
-
-        assertFalse(queue.isEmpty());
-
-        queue.clear();
-
-        // should be empty now
-        assertTrue(queue.isEmpty());
-    }
-
-    @Test
-    public void testIsEmpty() {
-        // test when empty
-        assertTrue(queue.isEmpty());
-
-        // all active
-        Forward msg1 = makeActive();
-        Forward msg2 = makeActive();
-        queue.add(msg1);
-        assertFalse(queue.isEmpty());
-
-        queue.add(msg2);
-        assertFalse(queue.isEmpty());
-
-        assertEquals(msg1, queue.poll());
-        assertFalse(queue.isEmpty());
-
-        assertEquals(msg2, queue.poll());
-        assertTrue(queue.isEmpty());
-
-        // active, expired, expired, active
-        queue.add(msg1);
-        queue.add(makeInactive());
-        queue.add(makeInactive());
-        queue.add(msg2);
-
-        assertEquals(msg1, queue.poll());
-        assertFalse(queue.isEmpty());
-
-        assertEquals(msg2, queue.poll());
-        assertTrue(queue.isEmpty());
-    }
-
-    @Test
-    public void testSize() {
-        queue = new EventQueue(2, 1000L);
-        assertEquals(0, queue.size());
-
-        queue.add(makeActive());
-        assertEquals(1, queue.size());
-
-        queue.poll();
-        assertEquals(0, queue.size());
-
-        queue.add(makeActive());
-        queue.add(makeActive());
-        assertEquals(2, queue.size());
-
-        queue.poll();
-        assertEquals(1, queue.size());
-
-        queue.poll();
-        assertEquals(0, queue.size());
-    }
-
-    @Test
-    public void testAdd() {
-        int nextra = 3;
-
-        // create excess messages
-        LinkedList<Forward> msgs = new LinkedList<>();
-        for (int x = 0; x < MAX_SIZE + nextra; ++x) {
-            msgs.add(makeActive());
-        }
-
-        // add them to the queue
-        msgs.forEach(msg -> queue.add(msg));
-
-        // should not have added too many messages
-        assertEquals(MAX_SIZE, queue.size());
-
-        // should have discarded the first "nextra" items
-        for (int x = 0; x < MAX_SIZE; ++x) {
-            assertEquals("x=" + x, msgs.get(x + nextra), queue.poll());
-        }
-
-        assertEquals(null, queue.poll());
-    }
-
-    @Test
-    public void testPoll() {
-        // poll when empty
-        assertNull(queue.poll());
-
-        // all active
-        Forward msg1 = makeActive();
-        Forward msg2 = makeActive();
-        queue.add(msg1);
-        queue.add(msg2);
-
-        assertEquals(msg1, queue.poll());
-        assertEquals(msg2, queue.poll());
-        assertEquals(null, queue.poll());
-
-        // active, expired, expired, active
-        queue.add(msg1);
-        queue.add(makeInactive());
-        queue.add(makeInactive());
-        queue.add(msg2);
-
-        assertEquals(msg1, queue.poll());
-        assertEquals(msg2, queue.poll());
-        assertEquals(null, queue.poll());
-
-        // one that's close to the age limit
-        msg1 = makeActive();
-        msg1.setCreateTimeMs(System.currentTimeMillis() - MAX_AGE_MS + 100);
-        queue.add(msg1);
-        assertEquals(msg1, queue.poll());
-        assertEquals(null, queue.poll());
-    }
-
-    private Forward makeActive() {
-        return new Forward(MY_SOURCE, MY_PROTO, MY_TOPIC, MY_PAYLOAD, MY_REQID);
-    }
-
-    private Forward makeInactive() {
-        Forward msg = new Forward(MY_SOURCE, MY_PROTO, MY_TOPIC, MY_PAYLOAD, MY_REQID);
-
-        msg.setCreateTimeMs(System.currentTimeMillis() - MAX_AGE_MS - 100);
-
-        return msg;
-    }
-
-}
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java
index 298af064..6280ebed 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/FeatureTest2.java
@@ -28,9 +28,11 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.onap.policy.drools.pooling.PoolingProperties.PREFIX;
 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.Properties;
 import java.util.TreeMap;
 import java.util.concurrent.CountDownLatch;
@@ -121,6 +123,7 @@ public class FeatureTest2 {
     // 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;
 
     /**
      * Sink for external DMaaP topic.
@@ -128,6 +131,11 @@ public class FeatureTest2 {
     private static TopicSink externalSink;
 
     /**
+     * Sink for internal DMaaP topic.
+     */
+    private static TopicSink internalSink;
+
+    /**
      * Context for the current test case.
      */
     private Context ctx;
@@ -137,18 +145,23 @@ public class FeatureTest2 {
     public static void setUpBeforeClass() {
         saveFeatureFactory = PoolingFeature.getFactory();
         saveManagerFactory = PoolingManagerImpl.getFactory();
+        saveDmaapFactory = DmaapManager.getFactory();
 
-        Properties props = makeSinkProperties(EXTERNAL_TOPIC);
-        externalSink = TopicEndpoint.manager.addTopicSinks(props).get(0);
+        externalSink = TopicEndpoint.manager.addTopicSinks(makeSinkProperties(EXTERNAL_TOPIC)).get(0);
         externalSink.start();
+
+        internalSink = TopicEndpoint.manager.addTopicSinks(makeSinkProperties(INTERNAL_TOPIC)).get(0);
+        internalSink.start();
     }
 
     @AfterClass
     public static void tearDownAfterClass() {
         PoolingFeature.setFactory(saveFeatureFactory);
         PoolingManagerImpl.setFactory(saveManagerFactory);
+        DmaapManager.setFactory(saveDmaapFactory);
 
         externalSink.stop();
+        internalSink.stop();
     }
 
     @Before
@@ -443,6 +456,7 @@ public class FeatureTest2 {
         private final AtomicBoolean sawMsg = new AtomicBoolean(false);
 
         private final TopicSource externalSource;
+        private final TopicSource internalSource;
 
         // mock objects
         private final PolicyEngine engine = mock(PolicyEngine.class);
@@ -458,8 +472,8 @@ public class FeatureTest2 {
             when(controller.getName()).thenReturn(CONTROLLER1);
             when(controller.getDrools()).thenReturn(drools);
 
-            Properties props = makeSourceProperties(EXTERNAL_TOPIC);
-            externalSource = TopicEndpoint.manager.addTopicSources(props).get(0);
+            externalSource = TopicEndpoint.manager.addTopicSources(makeSourceProperties(EXTERNAL_TOPIC)).get(0);
+            internalSource = TopicEndpoint.manager.addTopicSources(makeSourceProperties(INTERNAL_TOPIC)).get(0);
 
             // stop consuming events if the controller stops
             when(controller.stop()).thenAnswer(args -> {
@@ -489,6 +503,18 @@ public class FeatureTest2 {
          * "DMaaP" topic and its own internal "DMaaP" topic.
          */
         public void start() {
+            DmaapManager.setFactory(new DmaapManager.Factory() {
+                @Override
+                public List<TopicSource> getTopicSources() {
+                    return Arrays.asList(internalSource, externalSource);
+                }
+
+                @Override
+                public List<TopicSink> getTopicSinks() {
+                    return Arrays.asList(internalSink, externalSink);
+                }
+            });
+
             feature.beforeStart(engine);
             feature.afterCreate(controller);
 
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 64573ab0..d74b87f9 100644
--- a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java
+++ b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/PoolingManagerImplTest.java
@@ -23,7 +23,6 @@ package org.onap.policy.drools.pooling;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -106,7 +105,6 @@ public class PoolingManagerImplTest {
     private Properties plainProps;
     private PoolingProperties poolProps;
     private ListeningController controller;
-    private EventQueue eventQueue;
     private ClassExtractors extractors;
     private DmaapManager dmaap;
     private ScheduledThreadPoolExecutor sched;
@@ -146,14 +144,12 @@ public class PoolingManagerImplTest {
         active = new CountDownLatch(1);
 
         factory = mock(Factory.class);
-        eventQueue = mock(EventQueue.class);
         extractors = mock(ClassExtractors.class);
         dmaap = mock(DmaapManager.class);
         controller = mock(ListeningController.class);
         sched = mock(ScheduledThreadPoolExecutor.class);
         drools = mock(DroolsController.class);
 
-        when(factory.makeEventQueue(any())).thenReturn(eventQueue);
         when(factory.makeClassExtractors(any())).thenReturn(extractors);
         when(factory.makeDmaapManager(any())).thenReturn(dmaap);
         when(factory.makeScheduler()).thenReturn(sched);
@@ -304,24 +300,25 @@ public class PoolingManagerImplTest {
     @Test
     public void testBeforeStop() throws Exception {
         startMgr();
-        mgr.startDistributing(makeAssignments(true));
+        mgr.startDistributing(makeAssignments(false));
 
-        // verify that this message is not queued
         Forward msg = new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID);
         mgr.handle(msg);
-        verify(eventQueue, never()).add(msg);
+        verify(dmaap, times(START_PUB+1)).publish(any());
         
         mgr.beforeStop();
 
         verify(dmaap).stopConsumer(mgr);
         verify(sched).shutdownNow();
+        verify(dmaap, times(START_PUB+2)).publish(any());
         verify(dmaap).publish(contains("offline"));
 
         assertTrue(mgr.getCurrent() instanceof IdleState);
 
-        // verify that next message is queued
-        mgr.handle(msg);        
-        verify(eventQueue).add(msg);
+        // verify that next message is handled locally
+        mgr.handle(msg);
+        verify(dmaap, times(START_PUB+2)).publish(any());
+        verify(controller).onTopicEvent(CommInfrastructure.UEB, TOPIC2, THE_EVENT);
     }
 
     @Test
@@ -360,26 +357,8 @@ public class PoolingManagerImplTest {
         startMgr();
         mgr.beforeStop();
 
-        when(eventQueue.isEmpty()).thenReturn(false);
-        when(eventQueue.size()).thenReturn(3);
-
-        mgr.afterStop();
-
-        verify(eventQueue).clear();
-        verify(dmaap).stopPublisher(STD_OFFLINE_PUB_WAIT_MS);
-    }
-
-    @Test
-    public void testAfterStop_EmptyQueue() throws Exception {
-        startMgr();
-        mgr.beforeStop();
-
-        when(eventQueue.isEmpty()).thenReturn(true);
-        when(eventQueue.size()).thenReturn(0);
-
         mgr.afterStop();
 
-        verify(eventQueue, never()).clear();
         verify(dmaap).stopPublisher(STD_OFFLINE_PUB_WAIT_MS);
     }
 
@@ -499,18 +478,6 @@ public class PoolingManagerImplTest {
     }
 
     @Test
-    public void testInternalTopicFailed() throws Exception {
-        startMgr();
-
-        CountDownLatch latch = mgr.internalTopicFailed();
-
-        // wait for the thread to complete
-        assertTrue(latch.await(2, TimeUnit.SECONDS));
-
-        verify(controller).stop();
-    }
-
-    @Test
     public void testSchedule() throws Exception {
         // must start the scheduler
         startMgr();
@@ -693,14 +660,7 @@ public class PoolingManagerImplTest {
     public void testBeforeInsert_NoIntercept() throws Exception {
         startMgr();
 
-        long tbegin = System.currentTimeMillis();
-
-        assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-
-        ArgumentCaptor<Forward> msgCap = ArgumentCaptor.forClass(Forward.class);
-        verify(eventQueue).add(msgCap.capture());
-
-        validateMessageContent(tbegin, msgCap.getValue());
+        assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
     }
 
     @Test
@@ -726,37 +686,20 @@ public class PoolingManagerImplTest {
         startMgr();
 
         assertTrue(mgr.beforeInsert(null, TOPIC2, THE_EVENT, DECODED_EVENT));
-
-        // should not have tried to enqueue a message
-        verify(eventQueue, never()).add(any());
     }
 
     @Test
     public void testHandleExternalCommInfrastructureStringStringString() throws Exception {
         startMgr();
 
-        long tbegin = System.currentTimeMillis();
-
-        assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-
-        ArgumentCaptor<Forward> msgCap = ArgumentCaptor.forClass(Forward.class);
-        verify(eventQueue).add(msgCap.capture());
-
-        validateMessageContent(tbegin, msgCap.getValue());
+        assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
     }
 
     @Test
     public void testHandleExternalForward_NoAssignments() throws Exception {
         startMgr();
 
-        long tbegin = System.currentTimeMillis();
-
-        assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-
-        ArgumentCaptor<Forward> msgCap = ArgumentCaptor.forClass(Forward.class);
-        verify(eventQueue).add(msgCap.capture());
-
-        validateMessageContent(tbegin, msgCap.getValue());
+        assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
     }
 
     @Test
@@ -921,25 +864,26 @@ public class PoolingManagerImplTest {
     @Test
     public void testMakeForward() throws Exception {
         startMgr();
-
-        long tbegin = System.currentTimeMillis();
+        
+        // route the message to another host
+        mgr.startDistributing(makeAssignments(false));
 
         assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-
-        ArgumentCaptor<Forward> msgCap = ArgumentCaptor.forClass(Forward.class);
-        verify(eventQueue).add(msgCap.capture());
-
-        validateMessageContent(tbegin, msgCap.getValue());
+        
+        verify(dmaap, times(START_PUB+1)).publish(any());
     }
 
     @Test
     public void testMakeForward_InvalidMsg() throws Exception {
         startMgr();
+        
+        // route the message to another host
+        mgr.startDistributing(makeAssignments(false));
 
         assertTrue(mgr.beforeInsert(null, TOPIC2, THE_EVENT, DECODED_EVENT));
 
-        // should not have tried to enqueue a message
-        verify(eventQueue, never()).add(any());
+        // should not have tried to publish a message
+        verify(dmaap, times(START_PUB)).publish(any());
     }
 
     @Test
@@ -1064,69 +1008,27 @@ public class PoolingManagerImplTest {
         startMgr();
 
         // route the message to this host
-        assertNotNull(mgr.startDistributing(makeAssignments(true)));
+        mgr.startDistributing(makeAssignments(true));
         assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-        verify(eventQueue, never()).add(any());
+        verify(dmaap, times(START_PUB)).publish(any());
 
 
-        // null assignments should cause message to be queued
-        assertNull(mgr.startDistributing(null));
-        assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-        verify(eventQueue).add(any());
+        // null assignments should cause message to be processed locally
+        mgr.startDistributing(null);
+        assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
+        verify(dmaap, times(START_PUB)).publish(any());
 
 
         // route the message to this host
-        assertNotNull(mgr.startDistributing(makeAssignments(true)));
+        mgr.startDistributing(makeAssignments(true));
         assertFalse(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-        verify(eventQueue).add(any());
+        verify(dmaap, times(START_PUB)).publish(any());
 
 
         // route the message to the other host
-        assertNotNull(mgr.startDistributing(makeAssignments(false)));
+        mgr.startDistributing(makeAssignments(false));
         assertTrue(mgr.beforeInsert(CommInfrastructure.UEB, TOPIC2, THE_EVENT, DECODED_EVENT));
-        verify(eventQueue).add(any());
-    }
-
-    @Test
-    public void testStartDistributing_EventsInQueue_ProcessLocally() throws Exception {
-        startMgr();
-
-        // put items in the queue
-        LinkedList<Forward> lst = new LinkedList<>();
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-
-        when(eventQueue.poll()).thenAnswer(args -> lst.poll());
-
-        // route the messages to this host
-        CountDownLatch latch = mgr.startDistributing(makeAssignments(true));
-        assertTrue(latch.await(2, TimeUnit.SECONDS));
-
-        // all of the events should have been processed locally
-        verify(dmaap, times(START_PUB)).publish(any());
-        verify(controller, times(3)).onTopicEvent(CommInfrastructure.UEB, TOPIC2, THE_EVENT);
-    }
-
-    @Test
-    public void testStartDistributing_EventsInQueue_Forward() throws Exception {
-        startMgr();
-
-        // put items in the queue
-        LinkedList<Forward> lst = new LinkedList<>();
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-        lst.add(new Forward(mgr.getHost(), CommInfrastructure.UEB, TOPIC2, THE_EVENT, REQUEST_ID));
-
-        when(eventQueue.poll()).thenAnswer(args -> lst.poll());
-
-        // route the messages to the OTHER host
-        CountDownLatch latch = mgr.startDistributing(makeAssignments(false));
-        assertTrue(latch.await(2, TimeUnit.SECONDS));
-
-        // all of the events should have been forwarded
-        verify(dmaap, times(4)).publish(any());
-        verify(controller, never()).onTopicEvent(CommInfrastructure.UEB, TOPIC2, THE_EVENT);
+        verify(dmaap, times(START_PUB+1)).publish(any());
     }
 
     @Test
@@ -1228,22 +1130,6 @@ public class PoolingManagerImplTest {
     }
 
     /**
-     * Validates the message content.
-     * 
-     * @param tbegin creation time stamp must be no less than this
-     * @param msg message to be validated
-     */
-    private void validateMessageContent(long tbegin, Forward msg) {
-        assertEquals(0, msg.getNumHops());
-        assertTrue(msg.getCreateTimeMs() >= tbegin);
-        assertEquals(mgr.getHost(), msg.getSource());
-        assertEquals(CommInfrastructure.UEB, msg.getProtocol());
-        assertEquals(TOPIC2, msg.getTopic());
-        assertEquals(THE_EVENT, msg.getPayload());
-        assertEquals(REQUEST_ID, msg.getRequestId());
-    }
-
-    /**
      * 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.
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 27284dcd..f2701038 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
@@ -394,8 +394,8 @@ public class ActiveStateTest extends BasicStateTester {
         // fire the task - should transition
         assertEquals(next, task.third().fire());
 
-        // should stop distributing
-        verify(mgr).startDistributing(null);
+        // should continue to distribute
+        verify(mgr, never()).startDistributing(null);
 
         // should publish an offline message
         Offline msg = captureAdminMessage(Offline.class);
diff --git a/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java b/feature-pooling-dmaap/src/test/java/org/onap/policy/drools/pooling/state/QueryStateTest.java
index 80778ed4..7cd37581 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
@@ -257,8 +257,8 @@ public class QueryStateTest extends BasicStateTester {
 
         assertEquals(next, timer.second().fire());
 
-        // should stop distributing
-        verify(mgr).startDistributing(null);
+        // should continue distributing
+        verify(mgr, never()).startDistributing(null);
 
         Offline msg = captureAdminMessage(Offline.class);
         assertEquals(MY_HOST, msg.getSource());
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 01f49b55..18f12ff8 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
@@ -111,7 +111,7 @@ public class StartStateTest extends BasicStateTester {
 
         assertEquals(next, checker.second().fire());
 
-        verify(mgr).internalTopicFailed();
+        verify(mgr).startDistributing(null);
     }
 
     @Test
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 cdf9b59a..e3d383d9 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
@@ -423,7 +423,8 @@ public class StateTest extends BasicStateTester {
         State next2 = state.missedHeartbeat();
         assertEquals(next, next2);
 
-        verify(mgr).startDistributing(null);
+        // should continue to distribute
+        verify(mgr, never()).startDistributing(null);
 
         Offline msg = captureAdminMessage(Offline.class);
         assertEquals(MY_HOST, msg.getSource());
@@ -437,7 +438,8 @@ public class StateTest extends BasicStateTester {
         State next2 = state.internalTopicFailed();
         assertEquals(next, next2);
 
-        verify(mgr).internalTopicFailed();
+        // should stop distributing
+        verify(mgr).startDistributing(null);
 
         Offline msg = captureAdminMessage(Offline.class);
         assertEquals(MY_HOST, msg.getSource());