Add junit coverage to rest of policy-endpoint bus

Also extracted out common code from tests.
Fix checkstyle issues.
Be consistent in returning IllegalStateException when topic is not
found by any Topic Factory.
Added/updated some comments.
Use better name for "validate" argument.
Renamed test() to testDestroy().
Added NoopTopicPropertyBuilder.
Renamed BusTopicTestBase to TopicTestBase.

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

33 files changed, 1637 insertions, 648 deletions

diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
index 09078720..dc207a84 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactory.java
@@ -395,7 +395,7 @@ class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
             if (dmaapTopicWriters.containsKey(topic)) {
                 return dmaapTopicWriters.get(topic);
             } else {
-                throw new IllegalArgumentException("DmaapTopicSink for " + topic + " not found");
+                throw new IllegalStateException("DmaapTopicSink for " + topic + " not found");
             }
         }
     }
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
index f45164f8..ae6c6c3b 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactory.java
@@ -448,7 +448,7 @@ class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
             if (dmaapTopicSources.containsKey(topic)) {
                 return dmaapTopicSources.get(topic);
             } else {
-                throw new IllegalArgumentException("DmaapTopiceSource for " + topic + " not found");
+                throw new IllegalStateException("DmaapTopiceSource for " + topic + " not found");
             }
         }
     }
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactory.java
index d418bfac..b2c50184 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactory.java
@@ -101,7 +101,7 @@ class IndexedNoopTopicSinkFactory implements NoopTopicSinkFactory {
     /**
      * Logger.
      */
-    private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
+    private static Logger logger = LoggerFactory.getLogger(IndexedNoopTopicSinkFactory.class);
 
     /**
      * noop topic sinks map.
@@ -153,12 +153,8 @@ class IndexedNoopTopicSinkFactory implements NoopTopicSinkFactory {
     public NoopTopicSink build(List<String> servers, String topic, boolean managed) {
 
         List<String> noopSinkServers = servers;
-        if (noopSinkServers == null) {
-            noopSinkServers = new ArrayList<>();
-        }
-
-        if (noopSinkServers.isEmpty()) {
-            noopSinkServers.add("noop");
+        if (noopSinkServers == null || noopSinkServers.isEmpty()) {
+            noopSinkServers = Arrays.asList("noop");
         }
 
         if (topic == null || topic.isEmpty()) {
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactory.java
index bc107c1a..c200af5a 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactory.java
@@ -134,7 +134,7 @@ class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
                 return uebTopicSinks.get(busTopicParams.getTopic());
             }
 
-            UebTopicSink uebTopicWriter = new InlineUebTopicSink(busTopicParams);
+            UebTopicSink uebTopicWriter = makeSink(busTopicParams);
 
             if (busTopicParams.isManaged()) {
                 uebTopicSinks.put(busTopicParams.getTopic(), uebTopicWriter);
@@ -286,6 +286,16 @@ class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
         return new ArrayList<>(this.uebTopicSinks.values());
     }
 
+    /**
+     * Makes a new sink.
+     * 
+     * @param busTopicParams parameters to use to configure the sink
+     * @return a new sink
+     */
+    protected UebTopicSink makeSink(BusTopicParams busTopicParams) {
+        return new InlineUebTopicSink(busTopicParams);
+    }
+
 
     @Override
     public String toString() {
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactory.java
index f3f3e156..96315881 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactory.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactory.java
@@ -145,7 +145,7 @@ class IndexedUebTopicSourceFactory implements UebTopicSourceFactory {
                 return uebTopicSources.get(busTopicParams.getTopic());
             }
 
-            UebTopicSource uebTopicSource = new SingleThreadedUebTopicSource(busTopicParams);
+            UebTopicSource uebTopicSource = makeSource(busTopicParams);
 
             if (busTopicParams.isManaged()) {
                 uebTopicSources.put(busTopicParams.getTopic(), uebTopicSource);
@@ -284,6 +284,16 @@ class IndexedUebTopicSourceFactory implements UebTopicSourceFactory {
         return this.build(servers, topic, null, null);
     }
 
+    /**
+     * Makes a new source.
+     * 
+     * @param busTopicParams parameters to use to configure the source
+     * @return a new source
+     */
+    protected UebTopicSource makeSource(BusTopicParams busTopicParams) {
+        return new SingleThreadedUebTopicSource(busTopicParams);
+    }
+
     @Override
     public void destroy(String topic) {
 
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/BusTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/BusTopicFactoryTestBase.java
new file mode 100644
index 00000000..d68f55e6
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/BusTopicFactoryTestBase.java
@@ -0,0 +1,223 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+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.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.function.Function;
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+
+/**
+ * Base class for Topic Factory tests that use BusTopicParams.
+ *
+ * @param <T> type of topic managed by the factory
+ */
+public abstract class BusTopicFactoryTestBase<T extends Topic> extends TopicFactoryTestBase<T> {
+
+    /**
+     * Builds a topic.
+     *
+     * @param params the parameters used to configure the topic
+     * @return a new topic
+     */
+    protected abstract T buildTopic(BusTopicParams params);
+
+    /**
+     * Builds a topic.
+     *
+     * @param servers list of servers
+     * @param topic the topic name
+     * @return a new topic
+     */
+    protected abstract T buildTopic(List<String> servers, String topic);
+
+    /**
+     * Gets the parameters used to build the most recent topic.
+     *
+     * @return the most recent topic's parameters
+     */
+    protected abstract BusTopicParams getLastParams();
+
+    /**
+     * Tests building a topic using BusTopicParams.
+     */
+    public void testBuildBusTopicParams() {
+        initFactory();
+
+        // two unmanaged topics
+        T item = buildTopic(makeBuilder().managed(false).build());
+        T item2 = buildTopic(makeBuilder().managed(false).topic(TOPIC2).build());
+        assertNotNull(item);
+        assertNotNull(item2);
+        assertTrue(item != item2);
+
+        // duplicate topics, but since they aren't managed, they should be different
+        T item3 = buildTopic(makeBuilder().managed(false).build());
+        T item4 = buildTopic(makeBuilder().managed(false).build());
+        assertNotNull(item3);
+        assertNotNull(item4);
+        assertTrue(item != item3);
+        assertTrue(item != item4);
+        assertTrue(item3 != item4);
+
+        // two managed topics
+        T item5 = buildTopic(makeBuilder().build());
+        T item6 = buildTopic(makeBuilder().topic(TOPIC2).build());
+        assertNotNull(item5);
+        assertNotNull(item6);
+
+        // re-build same managed topics - should get exact same objects
+        assertTrue(item5 == buildTopic(makeBuilder().topic(MY_TOPIC).build()));
+        assertTrue(item6 == buildTopic(makeBuilder().topic(TOPIC2).build()));
+    }
+
+    /**
+     * Tests exception cases when building a topic using BusTopicParams.
+     */
+    public void testBuildBusTopicParams_Ex() {
+        // null topic
+        RuntimeException actual = expectException(() -> buildTopic(makeBuilder().topic(null).build()));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+
+        // empty topic
+        actual = expectException(() -> buildTopic(makeBuilder().topic("").build()));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+    }
+
+    /**
+     * Tests building a topic using a list of servers and a topic.
+     */
+    public void testBuildListOfStringString() {
+        initFactory();
+
+        T item1 = buildTopic(servers, MY_TOPIC);
+        assertNotNull(item1);
+
+        // check parameters that were used
+        BusTopicParams params = getLastParams();
+        assertEquals(servers, params.getServers());
+        assertEquals(MY_TOPIC, params.getTopic());
+        assertEquals(true, params.isManaged());
+        assertEquals(false, params.isUseHttps());
+
+        T item2 = buildTopic(servers, TOPIC2);
+        assertNotNull(item2);
+        assertTrue(item1 != item2);
+
+        // duplicate - should be the same, as these topics are managed
+        T item3 = buildTopic(servers, TOPIC2);
+        assertTrue(item2 == item3);
+    }
+
+    /**
+     * Tests building a topic using Properties. Verifies parameters specific to Bus
+     * topics.
+     */
+    public void testBuildProperties() {
+        initFactory();
+
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+
+        BusTopicParams params = getLastParams();
+        assertEquals(true, params.isManaged());
+        assertEquals(true, params.isUseHttps());
+        assertEquals(true, params.isAllowSelfSignedCerts());
+        assertEquals(MY_API_KEY, params.getApiKey());
+        assertEquals(MY_API_SECRET, params.getApiSecret());
+        assertEquals(Arrays.asList(SERVER), params.getServers());
+        assertEquals(MY_TOPIC, params.getTopic());
+    }
+
+    @Override
+    public void testBuildProperties_Variations() {
+        super.testBuildProperties_Variations();
+
+        // check boolean properties that default to true
+        checkDefault(PROPERTY_MANAGED_SUFFIX, BusTopicParams::isManaged);
+
+        // check boolean properties that default to false
+        checkDefault(PROPERTY_HTTP_HTTPS_SUFFIX, params -> !params.isUseHttps());
+        checkDefault(PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX, params -> !params.isAllowSelfSignedCerts());
+    }
+
+    /**
+     * Verifies that a parameter has the correct default, if the original builder property
+     * is not provided.
+     *
+     * @param builderName name of the builder property
+     * @param validate function to test the validity of the property
+     * @param values the values to which the property should be set, defaults to
+     *        {@code null} and ""
+     */
+    protected void checkDefault(String builderName, Function<BusTopicParams, Boolean> validate, Object... values) {
+        Object[] values2 = (values.length > 0 ? values : new Object[] {null, ""});
+
+        for (Object value : values2) {
+            // always start with a fresh factory
+            initFactory();
+
+            TopicPropertyBuilder builder = makePropBuilder().makeTopic(MY_TOPIC);
+
+            if (value == null) {
+                builder.removeTopicProperty(builderName);
+
+            } else {
+                builder.setTopicProperty(builderName, value.toString());
+            }
+
+            assertEquals("size for default " + value, 1, buildTopics(builder.build()).size());
+            assertTrue("default for " + value, validate.apply(getLastParams()));
+        }
+    }
+
+    /**
+     * Verifies that an "additional" property does not exist, if the original builder
+     * property is not provided.
+     *
+     * @param builderName name of the builder property
+     * @param addName name of the "additional" property
+     */
+    protected void expectNullAddProp(String builderName, String addName) {
+
+        // remove the property
+        initFactory();
+        Properties props = makePropBuilder().makeTopic(MY_TOPIC).removeTopicProperty(builderName).build();
+        assertEquals(1, buildTopics(props).size());
+        assertFalse(getLastParams().getAdditionalProps().containsKey(addName));
+
+
+        // repeat, this time using an empty string instead of null
+        initFactory();
+        props = makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(builderName, "").build();
+        assertEquals(1, buildTopics(props).size());
+        assertFalse(getLastParams().getAdditionalProps().containsKey(addName));
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java
new file mode 100644
index 00000000..c008a3bf
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicFactoryTestBase.java
@@ -0,0 +1,109 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX;
+
+import java.util.Map;
+import org.onap.policy.common.endpoints.event.comm.Topic;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+
+/**
+ * Base class for DmaapTopicXxxFactory tests.
+ *
+ * @param <T> type of topic managed by the factory
+ */
+public abstract class DmaapTopicFactoryTestBase<T extends Topic> extends BusTopicFactoryTestBase<T> {
+
+    public static final String MY_CONN_TIMEOUT = "200";
+    public static final String MY_READ_TIMEOUT = "201";
+    public static final String MY_ROUNDTRIP_TIMEOUT = "202";
+    public static final String MY_STICKINESS = "true";
+    public static final String MY_SUBCONTEXT = "my-subcontext";
+    public static final String MY_DME_VERSION = "my-version";
+
+    @Override
+    public void testBuildProperties() {
+        
+        super.testBuildProperties();
+        
+        // check properties specific to DMaaP/DME2
+        initFactory();
+
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+
+        BusTopicParams params = getLastParams();
+        assertEquals(MY_ENV, params.getEnvironment());
+        assertEquals(MY_LAT, params.getLatitude());
+        assertEquals(MY_LONG, params.getLongitude());
+        assertEquals(MY_PARTNER, params.getPartner());
+
+        Map<String, String> add = params.getAdditionalProps();
+        assertEquals(MY_CONN_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_EP_CONN_TIMEOUT_PROPERTY));
+        assertEquals(MY_READ_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_READ_TIMEOUT_PROPERTY));
+        assertEquals(MY_ROUNDTRIP_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY));
+        assertEquals(MY_ROUTE, add.get(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY));
+        assertEquals(MY_STICKINESS, add.get(DmaapTopicSinkFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY));
+        assertEquals(MY_SUBCONTEXT, add.get(DmaapTopicSinkFactory.DME2_SUBCONTEXT_PATH_PROPERTY));
+        assertEquals(MY_DME_VERSION, add.get(DmaapTopicSinkFactory.DME2_VERSION_PROPERTY));
+    }
+
+    @Override
+    public void testBuildProperties_Variations() {
+        super.testBuildProperties_Variations();
+
+        // check "additional" properties
+        expectNullAddProp(PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX,
+                        DmaapTopicSinkFactory.DME2_EP_CONN_TIMEOUT_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX,
+                        DmaapTopicSinkFactory.DME2_READ_TIMEOUT_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX,
+                        DmaapTopicSinkFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX, DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX,
+                        DmaapTopicSinkFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX,
+                        DmaapTopicSinkFactory.DME2_SUBCONTEXT_PATH_PROPERTY);
+
+        expectNullAddProp(PROPERTY_DMAAP_DME2_VERSION_SUFFIX, DmaapTopicSinkFactory.DME2_VERSION_PROPERTY);
+    }
+
+    @Override
+    public void testBuildListOfStringString() {
+        super.testBuildListOfStringString();
+
+        // check parameters that were used
+        BusTopicParams params = getLastParams();
+        assertEquals(false, params.isAllowSelfSignedCerts());
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
index 8ca22ebe..7276b445 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicPropertyBuilder.java
@@ -20,17 +20,19 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_AFT_ENV;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_API_KEY;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_API_SECRET;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_ENV;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_FETCH_LIMIT;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_FETCH_TIMEOUT;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_LAT;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_LONG;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_PARTITION;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_PARTNER;
-import static org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase.MY_ROUTE;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_AFT_ENV;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_KEY;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_SECRET;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_GROUP;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_INST;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_ENV;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_LIMIT;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_TIMEOUT;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_LAT;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_LONG;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_PARTITION;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_PARTNER;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_ROUTE;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX;
@@ -52,6 +54,8 @@ import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperti
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
 
@@ -87,6 +91,8 @@ public class DmaapTopicPropertyBuilder extends TopicPropertyBuilder {
     public DmaapTopicPropertyBuilder makeTopic(String topic) {
         addTopic(topic);
 
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX, MY_CONS_GROUP);
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX, MY_CONS_INST);
         setTopicProperty(PROPERTY_MANAGED_SUFFIX, "true");
         setTopicProperty(PROPERTY_HTTP_HTTPS_SUFFIX, "true");
         setTopicProperty(PROPERTY_TOPIC_AAF_MECHID_SUFFIX, MY_AAF_MECHID);
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactoryTest.java
index a9084764..b4c7fff8 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSinkFactoryTest.java
@@ -21,45 +21,19 @@
 package org.onap.policy.common.endpoints.event.comm.bus;
 
 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.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
 
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
-import java.util.function.Function;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
 
-public class DmaapTopicSinkFactoryTest extends BusTopicTestBase {
-
-    private static final String SERVER = "my-server";
-    private static final String TOPIC2 = "my-topic-2";
-
-    private static final String MY_CONN_TIMEOUT = "200";
-    private static final String MY_READ_TIMEOUT = "201";
-    private static final String MY_ROUNDTRIP_TIMEOUT = "202";
-    private static final String MY_STICKINESS = "true";
-    private static final String MY_SUBCONTEXT = "my-subcontext";
-    private static final String MY_DME_VERSION = "my-version";
+public class DmaapTopicSinkFactoryTest extends DmaapTopicFactoryTestBase<DmaapTopicSink> {
 
     private SinkFactory factory;
 
@@ -80,319 +54,106 @@ public class DmaapTopicSinkFactoryTest extends BusTopicTestBase {
 
     @Test
     public void testBuildBusTopicParams() {
-        // two unmanaged topics
-        DmaapTopicSink sink = factory.build(makeBuilder().managed(false).build());
-        DmaapTopicSink sink2 = factory.build(makeBuilder().managed(false).topic(TOPIC2).build());
-        assertNotNull(sink);
-        assertNotNull(sink2);
-        assertTrue(sink != sink2);
-
-        // duplicate topics, but since they aren't managed, they should be different
-        DmaapTopicSink sink3 = factory.build(makeBuilder().managed(false).build());
-        DmaapTopicSink sink4 = factory.build(makeBuilder().managed(false).build());
-        assertNotNull(sink3);
-        assertNotNull(sink4);
-        assertTrue(sink != sink3);
-        assertTrue(sink != sink4);
-        assertTrue(sink3 != sink4);
-
-        // two managed topics
-        DmaapTopicSink sink5 = factory.build(makeBuilder().build());
-        DmaapTopicSink sink6 = factory.build(makeBuilder().topic(TOPIC2).build());
-        assertNotNull(sink5);
-        assertNotNull(sink6);
-
-        // re-build same managed topics - should get exact same objects
-        assertTrue(sink5 == factory.build(BusTopicParams.builder().topic(MY_TOPIC).build()));
-        assertTrue(sink6 == factory.build(makeBuilder().topic(TOPIC2).build()));
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildBusTopicParams_NullTopic() {
-        factory.build(makeBuilder().topic(null).build());
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildBusTopicParams_EmptyTopic() {
-        factory.build(makeBuilder().topic("").build());
+        super.testBuildBusTopicParams();
+        super.testBuildBusTopicParams_Ex();
     }
 
     @Test
     public void testBuildListOfStringString() {
-        DmaapTopicSink sink1 = factory.build(servers, MY_TOPIC);
-        assertNotNull(sink1);
-
-        // check parameters that were used
-        BusTopicParams params = factory.params.get(0);
-        assertEquals(servers, params.getServers());
-        assertEquals(MY_TOPIC, params.getTopic());
-        assertEquals(true, params.isManaged());
-        assertEquals(false, params.isUseHttps());
-        assertEquals(false, params.isAllowSelfSignedCerts());
-
-        DmaapTopicSink sink2 = factory.build(servers, TOPIC2);
-        assertNotNull(sink2);
-        assertTrue(sink1 != sink2);
-
-        // duplicate - should be the same as these topics are managed
-        DmaapTopicSink sink3 = factory.build(Collections.emptyList(), TOPIC2);
-        assertTrue(sink2 == sink3);
+        super.testBuildListOfStringString();
     }
 
     @Test
     public void testBuildProperties() {
-        assertEquals(1, factory.build(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+        super.testBuildProperties();
+        super.testBuildProperties_Variations();
+        super.testBuildProperties_Multiple();
 
-        BusTopicParams params = factory.params.get(0);
-        assertEquals(true, params.isManaged());
-        assertEquals(true, params.isUseHttps());
-        assertEquals(true, params.isAllowSelfSignedCerts());
-        assertEquals(MY_API_KEY, params.getApiKey());
-        assertEquals(MY_API_SECRET, params.getApiSecret());
-        assertEquals(MY_ENV, params.getEnvironment());
-        assertEquals(MY_LAT, params.getLatitude());
-        assertEquals(MY_LONG, params.getLongitude());
+        // check sink-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
         assertEquals(MY_PARTITION, params.getPartitionId());
-        assertEquals(MY_PARTNER, params.getPartner());
-        assertEquals(Arrays.asList(SERVER), params.getServers());
-        assertEquals(MY_TOPIC, params.getTopic());
-
-        Map<String, String> add = params.getAdditionalProps();
-        assertEquals(MY_CONN_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_EP_CONN_TIMEOUT_PROPERTY));
-        assertEquals(MY_READ_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_READ_TIMEOUT_PROPERTY));
-        assertEquals(MY_ROUNDTRIP_TIMEOUT, add.get(DmaapTopicSinkFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY));
-        assertEquals(MY_ROUTE, add.get(DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY));
-        assertEquals(MY_STICKINESS, add.get(DmaapTopicSinkFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY));
-        assertEquals(MY_SUBCONTEXT, add.get(DmaapTopicSinkFactory.DME2_SUBCONTEXT_PATH_PROPERTY));
-        assertEquals(MY_DME_VERSION, add.get(DmaapTopicSinkFactory.DME2_VERSION_PROPERTY));
     }
 
     @Test
-    public void testBuildProperties_Variations() {
-        TopicPropertyBuilder builder = makePropBuilder().makeTopic(MY_TOPIC);
-
-        // null sinks
-        Properties props = builder.build();
-        props.remove(PROPERTY_DMAAP_SINK_TOPICS);
-        assertTrue(factory.build(props).isEmpty());
-
-        // empty sinks
-        props = builder.build();
-        props.setProperty(PROPERTY_DMAAP_SINK_TOPICS, "");
-        assertTrue(factory.build(props).isEmpty());
-
-        // null servers
-        assertTrue(factory.build(makePropBuilder().makeTopic(MY_TOPIC)
-                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).isEmpty());
-
-        // empty servers
-        assertTrue(factory.build(makePropBuilder().makeTopic(MY_TOPIC)
-                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).isEmpty());
-
-        // check boolean properties that default to true
-        checkDefault(builder, PROPERTY_MANAGED_SUFFIX, BusTopicParams::isManaged);
-
-        // check boolean properties that default to false
-        checkDefault(builder, PROPERTY_HTTP_HTTPS_SUFFIX, params -> ! params.isUseHttps());
-        checkDefault(builder, PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX,
-            params -> ! params.isAllowSelfSignedCerts());
-
-        // check "additional" properties
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_EP_CONN_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_READ_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_ROUTE_OFFER_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX,
-                        DmaapTopicSinkFactory.DME2_SUBCONTEXT_PATH_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_VERSION_SUFFIX, DmaapTopicSinkFactory.DME2_VERSION_PROPERTY);
+    public void testDestroyString_testGet_testInventory() {
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
     }
 
     @Test
-    public void testBuildProperties_Multiple() {
-        TopicPropertyBuilder builder =
-                        makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).addTopic(MY_TOPIC).addTopic(MY_TOPIC);
-
-        List<DmaapTopicSink> lst = factory.build(builder.build());
-        assertEquals(4, lst.size());
-
-        int index = 0;
-        DmaapTopicSink sink = lst.get(index++);
-        assertTrue(sink != lst.get(index++));
-        assertTrue(sink == lst.get(index++));
-        assertTrue(sink == lst.get(index++));
+    public void testDestroy() {
+        super.testDestroy();
     }
 
     @Test
-    public void testDestroyString_testGet_testInventory() {
-        List<DmaapTopicSink> lst = factory.build(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
-
-        int index = 0;
-        DmaapTopicSink sink1 = lst.get(index++);
-        DmaapTopicSink sink2 = lst.get(index++);
-
-        assertEquals(2, factory.inventory().size());
-        assertTrue(factory.inventory().contains(sink1));
-        assertTrue(factory.inventory().contains(sink2));
-
-        sink1.start();
-        sink2.start();
-
-        assertEquals(sink1, factory.get(MY_TOPIC));
-        assertEquals(sink2, factory.get(TOPIC2));
+    public void testGet() {
+        super.testGet_Ex();
+    }
 
-        factory.destroy(MY_TOPIC);
-        assertFalse(sink1.isAlive());
-        assertTrue(sink2.isAlive());
-        assertEquals(sink2, factory.get(TOPIC2));
-        assertEquals(1, factory.inventory().size());
-        assertTrue(factory.inventory().contains(sink2));
+    @Test
+    public void testToString() {
+        assertTrue(factory.toString().startsWith("IndexedDmaapTopicSinkFactory ["));
+    }
 
-        // repeat
-        factory.destroy(MY_TOPIC);
-        assertFalse(sink1.isAlive());
-        assertTrue(sink2.isAlive());
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
+            factory.destroy();
+        }
 
-        // with other topic
-        factory.destroy(TOPIC2);
-        assertFalse(sink1.isAlive());
-        assertFalse(sink2.isAlive());
-        assertEquals(0, factory.inventory().size());
+        factory = new SinkFactory();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDestroyString_NullTopic() {
-        factory.destroy(null);
+    @Override
+    protected List<DmaapTopicSink> buildTopics(Properties properties) {
+        return factory.build(properties);
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDestroyString_EmptyTopic() {
-        factory.destroy("");
+    @Override
+    protected DmaapTopicSink buildTopic(BusTopicParams params) {
+        return factory.build(params);
     }
 
-    @Test
-    public void testDestroy() {
-        List<DmaapTopicSink> lst = factory.build(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
-
-        int index = 0;
-        DmaapTopicSink sink1 = lst.get(index++);
-        DmaapTopicSink sink2 = lst.get(index++);
-
-        sink1.start();
-        sink2.start();
+    @Override
+    protected DmaapTopicSink buildTopic(List<String> servers, String topic) {
+        return factory.build(servers, topic);
+    }
 
+    @Override
+    protected void destroyFactory() {
         factory.destroy();
-        assertFalse(sink1.isAlive());
-        assertFalse(sink2.isAlive());
-        assertEquals(0, factory.inventory().size());
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_NullTopic() {
-        factory.get(null);
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_EmptyTopic() {
-        factory.get("");
+    @Override
+    protected List<DmaapTopicSink> getInventory() {
+        return factory.inventory();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_UnknownTopic() {
-        factory.build(makePropBuilder().makeTopic(MY_TOPIC).build());
-        factory.get(TOPIC2);
+    @Override
+    protected DmaapTopicSink getTopic(String topic) {
+        return factory.get(topic);
     }
 
-    @Test
-    public void testToString() {
-        assertTrue(factory.toString().startsWith("IndexedDmaapTopicSinkFactory ["));
+    @Override
+    protected BusTopicParams getLastParams() {
+        return factory.params.getLast();
     }
 
-    private DmaapTopicPropertyBuilder makePropBuilder() {
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
         return new DmaapTopicPropertyBuilder(PROPERTY_DMAAP_SINK_TOPICS);
     }
 
     /**
-     * Verifies that a parameter has the correct default, if the original builder property
-     * is not provided.
-     *
-     * @param builder used to build a set of properties
-     * @param builderName name of the builder property
-     * @param getter function to get the property from a set of parameters
-     */
-    private void checkDefault(TopicPropertyBuilder builder, String builderName,
-                    Function<BusTopicParams, Boolean> getter) {
-
-        /*
-         * Not sure why the "managed" property is treated differently, but it is.
-         */
-        String prefix = PROPERTY_DMAAP_SINK_TOPICS + "." + MY_TOPIC;
-
-        // always start with a fresh factory
-        factory.destroy();
-        factory = new SinkFactory();
-
-        Properties props = builder.build();
-        props.remove(prefix + builderName);
-
-        assertEquals(1, factory.build(props).size());
-        assertTrue(getter.apply(factory.params.get(0)));
-
-        // repeat, this time using an empty string instead of null
-        factory.destroy();
-        factory = new SinkFactory();
-
-        props.setProperty(prefix + builderName, "");
-
-        assertEquals(1, factory.build(props).size());
-        assertTrue(getter.apply(factory.params.get(0)));
-    }
-
-    /**
-     * Verifies that an "additional" property does not exist, if the original builder
-     * property is not provided.
-     *
-     * @param builder used to build a set of properties
-     * @param builderName name of the builder property
-     * @param addName name of the "additional" property
-     */
-    private void expectNullAddProp(TopicPropertyBuilder builder, String builderName, String addName) {
-        // always start with a fresh factory
-        factory.destroy();
-        factory = new SinkFactory();
-
-        Properties props = builder.build();
-        props.remove(PROPERTY_DMAAP_SINK_TOPICS + "." + MY_TOPIC + builderName);
-
-        assertEquals(1, factory.build(props).size());
-        assertFalse(factory.params.get(0).getAdditionalProps().containsKey(addName));
-
-        // repeat, this time using an empty string instead of null
-        factory.destroy();
-        factory = new SinkFactory();
-
-        props.setProperty(PROPERTY_DMAAP_SINK_TOPICS + "." + MY_TOPIC + builderName, "");
-
-        assertEquals(1, factory.build(props).size());
-        assertFalse(factory.params.get(0).getAdditionalProps().containsKey(addName));
-    }
-
-    /**
      * Factory that records the parameters of all of the sinks it creates.
      */
     private static class SinkFactory extends IndexedDmaapTopicSinkFactory {
-        private List<BusTopicParams> params = new LinkedList<>();
+        private Deque<BusTopicParams> params = new LinkedList<>();
 
         @Override
         protected DmaapTopicSink makeSink(BusTopicParams busTopicParams) {
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactoryTest.java
index cd276de5..ec6fcfcc 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactoryTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/DmaapTopicSourceFactoryTest.java
@@ -21,47 +21,22 @@
 package org.onap.policy.common.endpoints.event.comm.bus;
 
 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.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
-import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
 import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
 
-import java.util.Arrays;
-import java.util.Collections;
+import java.util.Deque;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
-import java.util.function.Function;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
 
-public class DmaapTopicSourceFactoryTest extends BusTopicTestBase {
-
-    private static final String SERVER = "my-server";
-    private static final String TOPIC2 = "my-topic-2";
-
-    private static final String MY_CONN_TIMEOUT = "200";
-    private static final String MY_READ_TIMEOUT = "201";
-    private static final String MY_ROUNDTRIP_TIMEOUT = "202";
-    private static final String MY_STICKINESS = "true";
-    private static final String MY_SUBCONTEXT = "my-subcontext";
-    private static final String MY_DME_VERSION = "my-version";
+public class DmaapTopicSourceFactoryTest extends DmaapTopicFactoryTestBase<DmaapTopicSource> {
 
     private SourceFactory factory;
 
@@ -82,145 +57,32 @@ public class DmaapTopicSourceFactoryTest extends BusTopicTestBase {
 
     @Test
     public void testBuildBusTopicParams() {
-        // two unmanaged topics
-        DmaapTopicSource source = factory.build(makeBuilder().managed(false).build());
-        DmaapTopicSource source2 = factory.build(makeBuilder().managed(false).topic(TOPIC2).build());
-        assertNotNull(source);
-        assertNotNull(source2);
-        assertTrue(source != source2);
-
-        // duplicate topics, but since they aren't managed, they should be different
-        DmaapTopicSource source3 = factory.build(makeBuilder().managed(false).build());
-        DmaapTopicSource source4 = factory.build(makeBuilder().managed(false).build());
-        assertNotNull(source3);
-        assertNotNull(source4);
-        assertTrue(source != source3);
-        assertTrue(source != source4);
-        assertTrue(source3 != source4);
-
-        // two managed topics
-        DmaapTopicSource source5 = factory.build(makeBuilder().build());
-        DmaapTopicSource source6 = factory.build(makeBuilder().topic(TOPIC2).build());
-        assertNotNull(source5);
-        assertNotNull(source6);
-
-        // re-build same managed topics - should get exact same objects
-        assertTrue(source5 == factory.build(BusTopicParams.builder().topic(MY_TOPIC).build()));
-        assertTrue(source6 == factory.build(makeBuilder().topic(TOPIC2).build()));
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildBusTopicParams_NullTopic() {
-        factory.build(makeBuilder().topic(null).build());
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testBuildBusTopicParams_EmptyTopic() {
-        factory.build(makeBuilder().topic("").build());
+        super.testBuildBusTopicParams();
+        super.testBuildBusTopicParams_Ex();
     }
 
     @Test
     public void testBuildProperties() {
-        assertEquals(1, factory.build(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+        super.testBuildProperties();
 
-        BusTopicParams params = factory.params.get(0);
-        assertEquals(true, params.isManaged());
-        assertEquals(true, params.isUseHttps());
-        assertEquals(true, params.isAllowSelfSignedCerts());
-        assertEquals(MY_API_KEY, params.getApiKey());
-        assertEquals(MY_API_SECRET, params.getApiSecret());
-        assertEquals(MY_ENV, params.getEnvironment());
-        assertEquals(MY_LAT, params.getLatitude());
-        assertEquals(MY_LONG, params.getLongitude());
-        assertEquals(MY_PARTNER, params.getPartner());
-        assertEquals(Arrays.asList(SERVER), params.getServers());
-        assertEquals(MY_TOPIC, params.getTopic());
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
+        assertEquals(MY_CONS_GROUP, params.getConsumerGroup());
+        assertEquals(MY_CONS_INST, params.getConsumerInstance());
         assertEquals(MY_FETCH_LIMIT, params.getFetchLimit());
         assertEquals(MY_FETCH_TIMEOUT, params.getFetchTimeout());
 
-        Map<String, String> add = params.getAdditionalProps();
-        assertEquals(MY_CONN_TIMEOUT, add.get(DmaapTopicSourceFactory.DME2_EP_CONN_TIMEOUT_PROPERTY));
-        assertEquals(MY_READ_TIMEOUT, add.get(DmaapTopicSourceFactory.DME2_READ_TIMEOUT_PROPERTY));
-        assertEquals(MY_ROUNDTRIP_TIMEOUT, add.get(DmaapTopicSourceFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY));
-        assertEquals(MY_ROUTE, add.get(DmaapTopicSourceFactory.DME2_ROUTE_OFFER_PROPERTY));
-        assertEquals(MY_STICKINESS, add.get(DmaapTopicSourceFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY));
-        assertEquals(MY_SUBCONTEXT, add.get(DmaapTopicSourceFactory.DME2_SUBCONTEXT_PATH_PROPERTY));
-        assertEquals(MY_DME_VERSION, add.get(DmaapTopicSourceFactory.DME2_VERSION_PROPERTY));
-    }
-
-    @Test
-    public void testBuildProperties_Variations() {
-        TopicPropertyBuilder builder = makePropBuilder().makeTopic(MY_TOPIC);
-
-        // null sources
-        Properties props = builder.build();
-        props.remove(PROPERTY_DMAAP_SOURCE_TOPICS);
-        assertTrue(factory.build(props).isEmpty());
-
-        // empty sources
-        props = builder.build();
-        props.setProperty(PROPERTY_DMAAP_SOURCE_TOPICS, "");
-        assertTrue(factory.build(props).isEmpty());
-
-        // null servers
-        assertTrue(factory.build(makePropBuilder().makeTopic(MY_TOPIC)
-                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).isEmpty());
-
-        // empty servers
-        assertTrue(factory.build(makePropBuilder().makeTopic(MY_TOPIC)
-                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).isEmpty());
-
-        // check boolean properties that default to true
-        checkDefault(builder, PROPERTY_MANAGED_SUFFIX, BusTopicParams::isManaged);
-
-        // check boolean properties that default to false
-        checkDefault(builder, PROPERTY_HTTP_HTTPS_SUFFIX, params -> !params.isUseHttps());
-        checkDefault(builder, PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX,
-            params -> !params.isAllowSelfSignedCerts());
+        super.testBuildProperties_Variations();
+        super.testBuildProperties_Multiple();
 
-        // check other properties having default values
-        checkDefault(builder, PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX,
-            params -> params.getFetchTimeout() == DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH, null, "",
-            "invalid-timeout");
-        checkDefault(builder, PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX,
-            params -> params.getFetchLimit() == DmaapTopicSource.DEFAULT_LIMIT_FETCH, null, "",
-            "invalid-limit");
-
-        // check "additional" properties
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_EP_CONN_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_READ_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_ROUNDTRIP_TIMEOUT_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_ROUTE_OFFER_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_SESSION_STICKINESS_REQUIRED_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX,
-                        DmaapTopicSourceFactory.DME2_SUBCONTEXT_PATH_PROPERTY);
-
-        expectNullAddProp(builder, PROPERTY_DMAAP_DME2_VERSION_SUFFIX, DmaapTopicSourceFactory.DME2_VERSION_PROPERTY);
-    }
-
-    @Test
-    public void testBuildProperties_Multiple() {
-        TopicPropertyBuilder builder =
-                        makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).addTopic(MY_TOPIC).addTopic(MY_TOPIC);
-
-        List<DmaapTopicSource> lst = factory.build(builder.build());
-        assertEquals(4, lst.size());
-
-        int index = 0;
-        DmaapTopicSource source = lst.get(index++);
-        assertTrue(source != lst.get(index++));
-        assertTrue(source == lst.get(index++));
-        assertTrue(source == lst.get(index++));
+        // check default values for source-specific parameters
+        checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX,
+            params2 -> params2.getFetchLimit() == DmaapTopicSource.DEFAULT_LIMIT_FETCH,
+            null, "", "invalid-limit-number");
+        
+        checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX,
+            params2 -> params2.getFetchTimeout() == DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH,
+            null, "", "invalid-timeout-number");
     }
 
     @Test
@@ -228,13 +90,8 @@ public class DmaapTopicSourceFactoryTest extends BusTopicTestBase {
         DmaapTopicSource source1 = factory.build(servers, MY_TOPIC, MY_API_KEY, MY_API_SECRET);
         assertNotNull(source1);
 
-        // check parameters that were used
-        BusTopicParams params = factory.params.get(0);
-        assertEquals(servers, params.getServers());
-        assertEquals(MY_TOPIC, params.getTopic());
-        assertEquals(true, params.isManaged());
-        assertEquals(false, params.isUseHttps());
-        assertEquals(false, params.isAllowSelfSignedCerts());
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
         assertEquals(MY_API_KEY, params.getApiKey());
         assertEquals(MY_API_SECRET, params.getApiSecret());
         assertEquals(DmaapTopicSource.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
@@ -243,108 +100,30 @@ public class DmaapTopicSourceFactoryTest extends BusTopicTestBase {
 
     @Test
     public void testBuildListOfStringString() {
-        DmaapTopicSource source1 = factory.build(servers, MY_TOPIC);
-        assertNotNull(source1);
+        super.testBuildListOfStringString();
 
-        // check parameters that were used
-        BusTopicParams params = factory.params.get(0);
-        assertEquals(servers, params.getServers());
-        assertEquals(MY_TOPIC, params.getTopic());
-        assertEquals(true, params.isManaged());
-        assertEquals(false, params.isUseHttps());
-        assertEquals(false, params.isAllowSelfSignedCerts());
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
         assertEquals(null, params.getApiKey());
         assertEquals(null, params.getApiSecret());
         assertEquals(DmaapTopicSource.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
         assertEquals(DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH, params.getFetchTimeout());
-
-        DmaapTopicSource source2 = factory.build(servers, TOPIC2);
-        assertNotNull(source2);
-        assertTrue(source1 != source2);
-
-        // duplicate - should be the same as these topics are managed
-        DmaapTopicSource source3 = factory.build(Collections.emptyList(), TOPIC2);
-        assertTrue(source2 == source3);
     }
 
     @Test
     public void testDestroyString_testGet_testInventory() {
-        List<DmaapTopicSource> lst = factory.build(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
-
-        int index = 0;
-        DmaapTopicSource source1 = lst.get(index++);
-        DmaapTopicSource source2 = lst.get(index++);
-
-        assertEquals(2, factory.inventory().size());
-        assertTrue(factory.inventory().contains(source1));
-        assertTrue(factory.inventory().contains(source2));
-
-        source1.start();
-        source2.start();
-
-        assertEquals(source1, factory.get(MY_TOPIC));
-        assertEquals(source2, factory.get(TOPIC2));
-
-        factory.destroy(MY_TOPIC);
-        assertFalse(source1.isAlive());
-        assertTrue(source2.isAlive());
-        assertEquals(source2, factory.get(TOPIC2));
-        assertEquals(1, factory.inventory().size());
-        assertTrue(factory.inventory().contains(source2));
-
-        // repeat
-        factory.destroy(MY_TOPIC);
-        assertFalse(source1.isAlive());
-        assertTrue(source2.isAlive());
-
-        // with other topic
-        factory.destroy(TOPIC2);
-        assertFalse(source1.isAlive());
-        assertFalse(source2.isAlive());
-        assertEquals(0, factory.inventory().size());
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testDestroyString_NullTopic() {
-        factory.destroy(null);
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testDestroyString_EmptyTopic() {
-        factory.destroy("");
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
     }
 
     @Test
     public void testDestroy() {
-        List<DmaapTopicSource> lst = factory.build(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
-
-        int index = 0;
-        DmaapTopicSource source1 = lst.get(index++);
-        DmaapTopicSource source2 = lst.get(index++);
-
-        source1.start();
-        source2.start();
-
-        factory.destroy();
-        assertFalse(source1.isAlive());
-        assertFalse(source2.isAlive());
-        assertEquals(0, factory.inventory().size());
+        super.testDestroy();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_NullTopic() {
-        factory.get(null);
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_EmptyTopic() {
-        factory.get("");
-    }
-
-    @Test(expected = IllegalArgumentException.class)
-    public void testGet_UnknownTopic() {
-        factory.build(makePropBuilder().makeTopic(MY_TOPIC).build());
-        factory.get(TOPIC2);
+    @Test
+    public void testGet() {
+        super.testGet_Ex();
     }
 
     @Test
@@ -352,75 +131,65 @@ public class DmaapTopicSourceFactoryTest extends BusTopicTestBase {
         assertTrue(factory.toString().startsWith("IndexedDmaapTopicSourceFactory ["));
     }
 
-    private DmaapTopicPropertyBuilder makePropBuilder() {
-        return new DmaapTopicPropertyBuilder(PROPERTY_DMAAP_SOURCE_TOPICS);
-    }
-
-    /**
-     * Verifies that a parameter has the correct default, if the original builder property
-     * is not provided.
-     *
-     * @param builder used to build a set of properties
-     * @param builderName name of the builder property
-     * @param getter function to get the property from a set of parameters
-     * @param values possible values to try, defaults to {null, ""}
-     */
-    private void checkDefault(TopicPropertyBuilder builder, String builderName,
-                    Function<BusTopicParams, Boolean> getter, Object... values) {
-
-        Object[] values2 = (values.length > 0 ? values : new String[] {null, ""});
-
-        for (Object value : values2) {
-            // always start with a fresh factory
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
             factory.destroy();
-            factory = new SourceFactory();
+        }
 
-            if (value == null) {
-                builder.removeTopicProperty(builderName);
+        factory = new SourceFactory();
+    }
 
-            } else {
-                builder.setTopicProperty(builderName, value.toString());
-            }
+    @Override
+    protected List<DmaapTopicSource> buildTopics(Properties properties) {
+        return factory.build(properties);
+    }
 
-            assertEquals(1, factory.build(builder.build()).size());
-            assertTrue(getter.apply(factory.params.get(0)));
-        }
+    @Override
+    protected DmaapTopicSource buildTopic(BusTopicParams params) {
+        return factory.build(params);
     }
 
-    /**
-     * Verifies that an "additional" property does not exist, if the original builder
-     * property is not provided.
-     *
-     * @param builder used to build a set of properties
-     * @param builderName name of the builder property
-     * @param addName name of the "additional" property
-     */
-    private void expectNullAddProp(TopicPropertyBuilder builder, String builderName, String addName) {
-        // always start with a fresh factory
+    @Override
+    protected DmaapTopicSource buildTopic(List<String> servers, String topic) {
+        return factory.build(servers, topic);
+    }
+
+    @Override
+    protected void destroyFactory() {
         factory.destroy();
-        factory = new SourceFactory();
+    }
 
-        Properties props = builder.build();
-        props.remove(PROPERTY_DMAAP_SOURCE_TOPICS + "." + MY_TOPIC + builderName);
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
+    }
 
-        assertEquals(1, factory.build(props).size());
-        assertFalse(factory.params.get(0).getAdditionalProps().containsKey(addName));
+    @Override
+    protected List<DmaapTopicSource> getInventory() {
+        return factory.inventory();
+    }
 
-        // repeat, this time using an empty string instead of null
-        factory.destroy();
-        factory = new SourceFactory();
+    @Override
+    protected DmaapTopicSource getTopic(String topic) {
+        return factory.get(topic);
+    }
 
-        props.setProperty(PROPERTY_DMAAP_SOURCE_TOPICS + "." + MY_TOPIC + builderName, "");
+    @Override
+    protected BusTopicParams getLastParams() {
+        return factory.params.getLast();
+    }
 
-        assertEquals(1, factory.build(props).size());
-        assertFalse(factory.params.get(0).getAdditionalProps().containsKey(addName));
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
+        return new DmaapTopicPropertyBuilder(PROPERTY_DMAAP_SOURCE_TOPICS);
     }
 
     /**
      * Factory that records the parameters of all of the sources it creates.
      */
     private static class SourceFactory extends IndexedDmaapTopicSourceFactory {
-        private List<BusTopicParams> params = new LinkedList<>();
+        private Deque<BusTopicParams> params = new LinkedList<>();
 
         @Override
         protected DmaapTopicSource makeSource(BusTopicParams busTopicParams) {
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java
new file mode 100644
index 00000000..ace51d93
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicPropertyBuilder.java
@@ -0,0 +1,58 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+
+public class NoopTopicPropertyBuilder extends TopicPropertyBuilder {
+
+    public static final String SERVER = "my-server";
+    public static final String TOPIC2 = "my-topic-2";
+
+    /**
+     * Constructs the object.
+     *
+     * @param prefix the prefix for the properties to be built
+     */
+    public NoopTopicPropertyBuilder(String prefix) {
+        super(prefix);
+    }
+
+    /**
+     * Adds a topic and configures it's properties with default values.
+     *
+     * @param topic the topic to be added
+     * @return this builder
+     */
+    public NoopTopicPropertyBuilder makeTopic(String topic) {
+        addTopic(topic);
+
+        setTopicProperty(PROPERTY_MANAGED_SUFFIX, "true");
+        setTopicProperty(PROPERTY_HTTP_HTTPS_SUFFIX, "true");
+        setTopicProperty(PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX, "true");
+        setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, SERVER);
+
+        return this;
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
new file mode 100644
index 00000000..2ba6fb3d
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkFactoryTest.java
@@ -0,0 +1,213 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class NoopTopicSinkFactoryTest extends TopicFactoryTestBase<NoopTopicSink> {
+
+    private static final List<String> NOOP_SERVERS = Arrays.asList("noop");
+
+    private IndexedNoopTopicSinkFactory factory;
+
+    /**
+     * Creates the object to be tested.
+     */
+    @Before
+    public void setUp() {
+        super.setUp();
+
+        factory = new IndexedNoopTopicSinkFactory();
+    }
+
+    @After
+    public void tearDown() {
+        factory.destroy();
+    }
+
+    @Test
+    public void testBuildListOfStringStringBoolean() {
+        initFactory();
+
+        NoopTopicSink item1 = buildTopic(servers, MY_TOPIC, true);
+        assertNotNull(item1);
+
+        assertEquals(servers, item1.getServers());
+        assertEquals(MY_TOPIC, item1.getTopic());
+
+        // managed topic - should not build a new one
+        assertEquals(item1, buildTopic(servers, MY_TOPIC, true));
+
+        NoopTopicSink item2 = buildTopic(servers, TOPIC2, true);
+        assertNotNull(item2);
+        assertTrue(item1 != item2);
+
+        // duplicate - should be the same, as these topics are managed
+        NoopTopicSink item3 = buildTopic(Collections.emptyList(), TOPIC2, true);
+        assertTrue(item2 == item3);
+
+        // null server list
+        initFactory();
+        assertEquals(NOOP_SERVERS, buildTopic(null, MY_TOPIC, true).getServers());
+
+        // empty server list
+        initFactory();
+        assertEquals(NOOP_SERVERS, buildTopic(Collections.emptyList(), MY_TOPIC, true).getServers());
+
+        // unmanaged topic
+        initFactory();
+        item1 = buildTopic(servers, MY_TOPIC, false);
+        assertTrue(item1 != buildTopic(servers, MY_TOPIC, false));
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testBuildListOfStringStringBoolean_NullTopic() {
+        buildTopic(servers, null, true);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void testBuildListOfStringStringBoolean_EmptyTopic() {
+        buildTopic(servers, "", true);
+    }
+
+    @Test
+    public void testBuildProperties() {
+        // managed topic
+        initFactory();
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        // unmanaged topic - get() will throw an exception
+        initFactory();
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
+                        .setTopicProperty(PROPERTY_MANAGED_SUFFIX, "false").build()).size());
+        assertNotNull(expectException(() -> factory.get(MY_TOPIC)));
+
+        // managed undefined - default to true
+        initFactory();
+        assertEquals(1, buildTopics(
+                        makePropBuilder().makeTopic(MY_TOPIC).removeTopicProperty(PROPERTY_MANAGED_SUFFIX).build())
+                                        .size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        // managed empty - default to true
+        initFactory();
+        assertEquals(1, buildTopics(
+                        makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_MANAGED_SUFFIX, "").build())
+                                        .size());
+        assertNotNull(factory.get(MY_TOPIC));
+
+        initFactory();
+
+        // null topic list
+        assertTrue(buildTopics(makePropBuilder().build()).isEmpty());
+
+        // empty topic list
+        assertTrue(buildTopics(makePropBuilder().addTopic("").build()).isEmpty());
+
+        // null server list
+        initFactory();
+        NoopTopicSink sink = buildTopics(makePropBuilder().makeTopic(MY_TOPIC)
+                        .removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX).build()).get(0);
+        assertEquals(NOOP_SERVERS, sink.getServers());
+
+        // empty server list
+        initFactory();
+        sink = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, "")
+                        .build()).get(0);
+        assertEquals(NOOP_SERVERS, sink.getServers());
+
+        // test other options
+        super.testBuildProperties_Multiple();
+    }
+
+    @Test
+    public void testDestroyString_testGet_testInventory() {
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
+    }
+
+    @Test
+    public void testDestroy() {
+        super.testDestroy();
+    }
+
+    @Test
+    public void testGet() {
+        super.testGet_Ex();
+    }
+
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
+            factory.destroy();
+        }
+
+        factory = new IndexedNoopTopicSinkFactory();
+    }
+
+    @Override
+    protected List<NoopTopicSink> buildTopics(Properties properties) {
+        return factory.build(properties);
+    }
+
+    protected NoopTopicSink buildTopic(List<String> servers, String topic, boolean managed) {
+        return factory.build(servers, topic, managed);
+    }
+
+    @Override
+    protected void destroyFactory() {
+        factory.destroy();
+    }
+
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
+    }
+
+    @Override
+    protected List<NoopTopicSink> getInventory() {
+        return factory.inventory();
+    }
+
+    @Override
+    protected NoopTopicSink getTopic(String topic) {
+        return factory.get(topic);
+    }
+
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
+        return new NoopTopicPropertyBuilder(PROPERTY_NOOP_SINK_TOPICS);
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
index 63c29111..8a5b7b2e 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/NoopTopicSinkTest.java
@@ -32,7 +32,7 @@ import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicListener;
 
-public class NoopTopicSinkTest extends BusTopicTestBase {
+public class NoopTopicSinkTest extends TopicTestBase {
 
     private NoopTopicSink sink;
 
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicFactoryTestBase.java
new file mode 100644
index 00000000..29816c6e
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicFactoryTestBase.java
@@ -0,0 +1,245 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+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.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+
+import java.util.List;
+import java.util.Properties;
+import org.onap.policy.common.endpoints.event.comm.Topic;
+
+/**
+ * Base class for XxxTopicFactory tests.
+ *
+ * @param <T> type of topic managed by the factory
+ */
+public abstract class TopicFactoryTestBase<T extends Topic> extends TopicTestBase {
+
+    public static final String SERVER = "my-server";
+    public static final String TOPIC2 = "my-topic-2";
+
+    /**
+     * Initializes a new factory.
+     */
+    protected abstract void initFactory();
+
+    /**
+     * Makes a property builder.
+     *
+     * @return a new property builder
+     */
+    protected abstract TopicPropertyBuilder makePropBuilder();
+
+    /**
+     * Builds a set of topics.
+     *
+     * @param properties the properties used to configure the topics
+     * @return a list of new topics
+     */
+    protected abstract List<T> buildTopics(Properties properties);
+
+    /**
+     * Destroys the factory.
+     */
+    protected abstract void destroyFactory();
+
+    /**
+     * Destroys a topic within the factory.
+     *
+     * @param topic the topic to destroy
+     */
+    protected abstract void destroyTopic(String topic);
+
+    /**
+     * Gets the list of topics from the factory.
+     *
+     * @return the topic inventory
+     */
+    protected abstract List<T> getInventory();
+
+    /**
+     * Gets a topic from the factory.
+     *
+     * @param topic the topic name
+     * @return the topic
+     */
+    protected abstract T getTopic(String topic);
+
+
+    /**
+     * Tests building a topic using varied Properties.
+     */
+    public void testBuildProperties_Variations() {
+        initFactory();
+
+        // null topic list
+        assertTrue(buildTopics(makePropBuilder().build()).isEmpty());
+
+        // empty topic list
+        assertTrue(buildTopics(makePropBuilder().addTopic("").build()).isEmpty());
+
+        // null servers
+        assertTrue(buildTopics(makePropBuilder().makeTopic(MY_TOPIC).removeTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX)
+                        .build()).isEmpty());
+
+        // empty servers
+        assertTrue(buildTopics(makePropBuilder().makeTopic(MY_TOPIC).setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, "")
+                        .build()).isEmpty());
+    }
+
+    /**
+     * Tests building multiple topics using Properties.
+     */
+    public void testBuildProperties_Multiple() {
+        initFactory();
+
+        // make two fully-defined topics, and add two duplicate topic names to the list
+        TopicPropertyBuilder builder =
+                        makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).addTopic(MY_TOPIC).addTopic(MY_TOPIC);
+
+        List<T> lst = buildTopics(builder.build());
+        assertEquals(4, lst.size());
+
+        int index = 0;
+        T item = lst.get(index++);
+        assertTrue(item != lst.get(index++));
+        assertTrue(item == lst.get(index++));
+        assertTrue(item == lst.get(index++));
+    }
+
+    /**
+     * Tests destroy(topic), get(topic), and inventory() methods.
+     */
+    public void testDestroyString_testGet_testInventory() {
+        initFactory();
+
+        List<T> lst = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
+
+        int index = 0;
+        T item1 = lst.get(index++);
+        T item2 = lst.get(index++);
+
+        assertEquals(2, getInventory().size());
+        assertTrue(getInventory().contains(item1));
+        assertTrue(getInventory().contains(item2));
+
+        item1.start();
+        item2.start();
+
+        assertEquals(item1, getTopic(MY_TOPIC));
+        assertEquals(item2, getTopic(TOPIC2));
+
+        destroyTopic(MY_TOPIC);
+        assertFalse(item1.isAlive());
+        assertTrue(item2.isAlive());
+        assertEquals(item2, getTopic(TOPIC2));
+        assertEquals(1, getInventory().size());
+        assertTrue(getInventory().contains(item2));
+
+        // repeat
+        destroyTopic(MY_TOPIC);
+        assertFalse(item1.isAlive());
+        assertTrue(item2.isAlive());
+
+        // with other topic
+        destroyTopic(TOPIC2);
+        assertFalse(item1.isAlive());
+        assertFalse(item2.isAlive());
+        assertEquals(0, getInventory().size());
+    }
+
+    /**
+     * Tests exception cases with destroy(topic).
+     */
+    public void testDestroyString_Ex() {
+        // null topic
+        RuntimeException actual = expectException(() -> destroyTopic(null));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+
+        // empty topic
+        actual = expectException(() -> destroyTopic(""));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+    }
+
+    /**
+     * Tests the destroy() method.
+     */
+    public void testDestroy() {
+        initFactory();
+
+        List<T> lst = buildTopics(makePropBuilder().makeTopic(MY_TOPIC).makeTopic(TOPIC2).build());
+
+        int index = 0;
+        T item1 = lst.get(index++);
+        T item2 = lst.get(index++);
+
+        item1.start();
+        item2.start();
+
+        destroyFactory();
+
+        assertFalse(item1.isAlive());
+        assertFalse(item2.isAlive());
+        assertEquals(0, getInventory().size());
+    }
+
+    /**
+     * Tests exception cases with get(topic).
+     */
+    public void testGet_Ex() {
+        // null topic
+        RuntimeException actual = expectException(() -> getTopic(null));
+        assertEquals("null topic", IllegalArgumentException.class, actual.getClass());
+
+        // empty topic
+        actual = expectException(() -> getTopic(""));
+        assertEquals("empty topic", IllegalArgumentException.class, actual.getClass());
+
+        // unknown topic
+        initFactory();
+        buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build());
+
+        actual = expectException(() -> getTopic(TOPIC2));
+        assertEquals("unknown topic", IllegalStateException.class, actual.getClass());
+    }
+
+    /**
+     * Runs a function that is expected to throw an exception. Invokes fail() if the
+     * function does not throw an exception.
+     *
+     * @param function the function to run
+     * @return the exception thrown by the function
+     */
+    public RuntimeException expectException(Runnable function) {
+        try {
+            function.run();
+            fail("missing exception");
+            return null;
+
+        } catch (RuntimeException e) {
+            return e;
+        }
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicPropertyBuilder.java
index 4982d11d..e8031c1a 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicPropertyBuilder.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicPropertyBuilder.java
@@ -25,7 +25,7 @@ import java.util.Properties;
 /**
  * Builder of properties used when configuring topics.
  */
-public class TopicPropertyBuilder {
+public abstract class TopicPropertyBuilder {
     private final Properties properties = new Properties();
     private final String prefix;
     private String topicPrefix;
@@ -37,7 +37,6 @@ public class TopicPropertyBuilder {
      */
     public TopicPropertyBuilder(String prefix) {
         this.prefix = prefix;
-        properties.setProperty(prefix, "");
     }
 
     /**
@@ -53,6 +52,15 @@ public class TopicPropertyBuilder {
     }
 
     /**
+     * Adds a topic to the list of topics, configuring all of its properties with default
+     * values.
+     * 
+     * @param topic the topic to be added
+     * @return this builder
+     */
+    public abstract TopicPropertyBuilder makeTopic(String topic);
+
+    /**
      * Adds a topic to the list of topics. Also sets the current topic so that subsequent
      * invocations of property methods will manipulate the topic's properties.
      *
@@ -62,10 +70,12 @@ public class TopicPropertyBuilder {
     public TopicPropertyBuilder addTopic(String topic) {
         // add topic to the list of topics
         String topicList = properties.getProperty(prefix);
-        if (!topicList.isEmpty()) {
-            topicList += ",";
+        if (topicList == null || topicList.isEmpty()) {
+            topicList = topic;
+        } else {
+            topicList += "," + topic;
         }
-        topicList += topic;
+
         properties.setProperty(prefix, topicList);
 
         setTopic(topic);
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/BusTopicTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicTestBase.java
index ba52f191..fb94e53e 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/BusTopicTestBase.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/TopicTestBase.java
@@ -28,9 +28,9 @@ import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams.TopicParamsBuilder;
 
 /**
- * Base class for BusTopicXxxTest classes.
+ * Base class for Topic Test classes.
  */
-public class BusTopicTestBase {
+public class TopicTestBase {
 
     public static final String MY_AFT_ENV = "my-aft-env";
     public static final String MY_API_KEY = "my-api-key";
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java
new file mode 100644
index 00000000..fbc2e46b
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicFactoryTestBase.java
@@ -0,0 +1,48 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collections;
+import org.onap.policy.common.endpoints.event.comm.Topic;
+
+/**
+ * Base class for UebTopicXxxFactory tests.
+ *
+ * @param <T> type of topic managed by the factory
+ */
+public abstract class UebTopicFactoryTestBase<T extends Topic> extends BusTopicFactoryTestBase<T> {
+
+    @Override
+    public void testBuildBusTopicParams_Ex() {
+
+        super.testBuildBusTopicParams_Ex();
+
+        // null servers
+        RuntimeException actual = expectException(() -> buildTopic(makeBuilder().servers(null).build()));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+
+        // empty servers
+        actual = expectException(() -> buildTopic(makeBuilder().servers(Collections.emptyList()).build()));
+        assertEquals(IllegalArgumentException.class, actual.getClass());
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
new file mode 100644
index 00000000..efa1f125
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicPropertyBuilder.java
@@ -0,0 +1,89 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_AFT_ENV;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_KEY;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_API_SECRET;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_GROUP;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_CONS_INST;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_LIMIT;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_FETCH_TIMEOUT;
+import static org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase.MY_PARTITION;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
+
+public class UebTopicPropertyBuilder extends TopicPropertyBuilder {
+
+    public static final String SERVER = "my-server";
+    public static final String TOPIC2 = "my-topic-2";
+
+    public static final String MY_AAF_MECHID = "my-aaf-mechid";
+    public static final String MY_AAF_PASSWD = "my-aaf-passwd";
+
+    /**
+     * Constructs the object.
+     *
+     * @param prefix the prefix for the properties to be built
+     */
+    public UebTopicPropertyBuilder(String prefix) {
+        super(prefix);
+    }
+
+    /**
+     * Adds a topic and configures it's properties with default values.
+     *
+     * @param topic the topic to be added
+     * @return this builder
+     */
+    public UebTopicPropertyBuilder makeTopic(String topic) {
+        addTopic(topic);
+
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX, MY_CONS_GROUP);
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX, MY_CONS_INST);
+        setTopicProperty(PROPERTY_MANAGED_SUFFIX, "true");
+        setTopicProperty(PROPERTY_HTTP_HTTPS_SUFFIX, "true");
+        setTopicProperty(PROPERTY_TOPIC_AAF_MECHID_SUFFIX, MY_AAF_MECHID);
+        setTopicProperty(PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX, MY_AAF_PASSWD);
+        setTopicProperty(PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX, MY_AFT_ENV);
+        setTopicProperty(PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX, "true");
+        setTopicProperty(PROPERTY_TOPIC_API_KEY_SUFFIX, MY_API_KEY);
+        setTopicProperty(PROPERTY_TOPIC_API_SECRET_SUFFIX, MY_API_SECRET);
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX, MY_FETCH_LIMIT);
+        setTopicProperty(PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX, MY_FETCH_TIMEOUT);
+        setTopicProperty(PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX, MY_PARTITION);
+        setTopicProperty(PROPERTY_TOPIC_SERVERS_SUFFIX, SERVER);
+
+        return this;
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java
new file mode 100644
index 00000000..10d61f60
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkFactoryTest.java
@@ -0,0 +1,171 @@
+/*
+ * ============LICENSE_START=======================================================
+ * policy-endpoints
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+
+public class UebTopicSinkFactoryTest extends UebTopicFactoryTestBase<UebTopicSink> {
+
+    private SinkFactory factory;
+
+    /**
+     * Creates the object to be tested.
+     */
+    @Before
+    public void setUp() {
+        super.setUp();
+
+        factory = new SinkFactory();
+    }
+
+    @After
+    public void tearDown() {
+        factory.destroy();
+    }
+
+    @Test
+    public void testBuildBusTopicParams() {
+        super.testBuildBusTopicParams();
+        super.testBuildBusTopicParams_Ex();
+    }
+
+    @Test
+    public void testBuildListOfStringString() {
+        super.testBuildListOfStringString();
+
+        // check parameters that were used
+        BusTopicParams params = getLastParams();
+        assertEquals(false, params.isAllowSelfSignedCerts());
+    }
+
+    @Test
+    public void testBuildProperties() {
+        super.testBuildProperties();
+        super.testBuildProperties_Variations();
+        super.testBuildProperties_Multiple();
+        
+        initFactory();
+
+        assertEquals(1, buildTopics(makePropBuilder().makeTopic(MY_TOPIC).build()).size());
+
+        BusTopicParams params = getLastParams();
+        assertEquals(MY_PARTITION, params.getPartitionId());
+    }
+
+    @Test
+    public void testDestroyString_testGet_testInventory() {
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
+    }
+
+    @Test
+    public void testDestroy() {
+        super.testDestroy();
+    }
+
+    @Test
+    public void testGet() {
+        super.testGet_Ex();
+    }
+
+    @Test
+    public void testToString() {
+        assertTrue(factory.toString().startsWith("IndexedUebTopicSinkFactory ["));
+    }
+
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
+            factory.destroy();
+        }
+
+        factory = new SinkFactory();
+    }
+
+    @Override
+    protected List<UebTopicSink> buildTopics(Properties properties) {
+        return factory.build(properties);
+    }
+
+    @Override
+    protected UebTopicSink buildTopic(BusTopicParams params) {
+        return factory.build(params);
+    }
+
+    @Override
+    protected UebTopicSink buildTopic(List<String> servers, String topic) {
+        return factory.build(servers, topic);
+    }
+
+    @Override
+    protected void destroyFactory() {
+        factory.destroy();
+    }
+
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
+    }
+
+    @Override
+    protected List<UebTopicSink> getInventory() {
+        return factory.inventory();
+    }
+
+    @Override
+    protected UebTopicSink getTopic(String topic) {
+        return factory.get(topic);
+    }
+
+    @Override
+    protected BusTopicParams getLastParams() {
+        return factory.params.getLast();
+    }
+
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
+        return new UebTopicPropertyBuilder(PROPERTY_UEB_SINK_TOPICS);
+    }
+
+    /**
+     * Factory that records the parameters of all of the sinks it creates.
+     */
+    private static class SinkFactory extends IndexedUebTopicSinkFactory {
+        private Deque<BusTopicParams> params = new LinkedList<>();
+
+        @Override
+        protected UebTopicSink makeSink(BusTopicParams busTopicParams) {
+            params.add(busTopicParams);
+            return super.makeSink(busTopicParams);
+        }
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java
new file mode 100644
index 00000000..3e7aca01
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSinkTest.java
@@ -0,0 +1,34 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+
+public class UebTopicSinkTest {
+
+    @Test
+    public void test() {
+        assertNotNull(UebTopicSink.factory);
+    }
+
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java
new file mode 100644
index 00000000..73e3930d
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceFactoryTest.java
@@ -0,0 +1,203 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX;
+import static org.onap.policy.common.endpoints.properties.PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+
+public class UebTopicSourceFactoryTest extends UebTopicFactoryTestBase<UebTopicSource> {
+
+    private SourceFactory factory;
+
+    /**
+     * Creates the object to be tested.
+     */
+    @Before
+    public void setUp() {
+        super.setUp();
+
+        factory = new SourceFactory();
+    }
+
+    @After
+    public void tearDown() {
+        factory.destroy();
+    }
+
+    @Test
+    public void testBuildBusTopicParams() {
+        super.testBuildBusTopicParams();
+        super.testBuildBusTopicParams_Ex();
+    }
+
+    @Test
+    public void testBuildProperties() {
+        
+        super.testBuildProperties();
+
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
+        assertEquals(MY_CONS_GROUP, params.getConsumerGroup());
+        assertEquals(MY_CONS_INST, params.getConsumerInstance());
+        assertEquals(MY_FETCH_LIMIT, params.getFetchLimit());
+        assertEquals(MY_FETCH_TIMEOUT, params.getFetchTimeout());
+        
+        super.testBuildProperties_Variations();
+        super.testBuildProperties_Multiple();
+
+        // check default values for source-specific parameters
+        checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX,
+            params2 -> params2.getFetchLimit() == UebTopicSource.DEFAULT_LIMIT_FETCH,
+            null, "", "invalid-limit-number");
+        
+        checkDefault(PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX,
+            params2 -> params2.getFetchTimeout() == UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH,
+            null, "", "invalid-timeout-number");
+    }
+
+    @Test
+    public void testBuildListOfStringStringStringString() {
+        UebTopicSource source1 = factory.build(servers, MY_TOPIC, MY_API_KEY, MY_API_SECRET);
+        assertNotNull(source1);
+
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
+        assertEquals(MY_API_KEY, params.getApiKey());
+        assertEquals(MY_API_SECRET, params.getApiSecret());
+        assertEquals(UebTopicSource.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
+        assertEquals(UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH, params.getFetchTimeout());
+    }
+
+    @Test
+    public void testBuildListOfStringString() {
+        super.testBuildListOfStringString();
+
+        // check source-specific parameters that were used
+        BusTopicParams params = factory.params.getFirst();
+        assertEquals(null, params.getApiKey());
+        assertEquals(null, params.getApiSecret());
+        assertEquals(UebTopicSource.DEFAULT_LIMIT_FETCH, params.getFetchLimit());
+        assertEquals(UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH, params.getFetchTimeout());
+
+        assertEquals(true, params.isAllowSelfSignedCerts());
+    }
+
+    @Test
+    public void testDestroyString_testGet_testInventory() {
+        super.testDestroyString_testGet_testInventory();
+        super.testDestroyString_Ex();
+    }
+
+    @Test
+    public void testDestroy() {
+        super.testDestroy();
+    }
+
+    @Test
+    public void testGet() {
+        super.testGet_Ex();
+    }
+
+    @Test
+    public void testToString() {
+        assertTrue(factory.toString().startsWith("IndexedUebTopicSourceFactory ["));
+    }
+
+    @Override
+    protected void initFactory() {
+        if (factory != null) {
+            factory.destroy();
+        }
+
+        factory = new SourceFactory();
+    }
+
+    @Override
+    protected List<UebTopicSource> buildTopics(Properties properties) {
+        return factory.build(properties);
+    }
+
+    @Override
+    protected UebTopicSource buildTopic(BusTopicParams params) {
+        return factory.build(params);
+    }
+
+    @Override
+    protected UebTopicSource buildTopic(List<String> servers, String topic) {
+        return factory.build(servers, topic);
+    }
+
+    @Override
+    protected void destroyFactory() {
+        factory.destroy();
+    }
+
+    @Override
+    protected void destroyTopic(String topic) {
+        factory.destroy(topic);
+    }
+
+    @Override
+    protected List<UebTopicSource> getInventory() {
+        return factory.inventory();
+    }
+
+    @Override
+    protected UebTopicSource getTopic(String topic) {
+        return factory.get(topic);
+    }
+
+    @Override
+    protected BusTopicParams getLastParams() {
+        return factory.params.getLast();
+    }
+
+    @Override
+    protected TopicPropertyBuilder makePropBuilder() {
+        return new UebTopicPropertyBuilder(PROPERTY_UEB_SOURCE_TOPICS);
+    }
+
+    /**
+     * Factory that records the parameters of all of the sources it creates.
+     */
+    private static class SourceFactory extends IndexedUebTopicSourceFactory {
+        private Deque<BusTopicParams> params = new LinkedList<>();
+
+        @Override
+        protected UebTopicSource makeSource(BusTopicParams busTopicParams) {
+            params.add(busTopicParams);
+            return super.makeSource(busTopicParams);
+        }
+    }
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java
new file mode 100644
index 00000000..11bb3f4d
--- /dev/null
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/UebTopicSourceTest.java
@@ -0,0 +1,34 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * 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.common.endpoints.event.comm.bus;
+
+import static org.junit.Assert.assertNotNull;
+
+import org.junit.Test;
+
+public class UebTopicSourceTest {
+
+    @Test
+    public void test() {
+        assertNotNull(UebTopicSource.factory);
+    }
+
+}
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
index ef4d5a00..24dc9e10 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
@@ -37,14 +37,14 @@ import java.util.Collections;
 import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.CambriaConsumerWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapAafConsumerWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapConsumerWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapDmeConsumerWrapper;
 import org.powermock.reflect.Whitebox;
 
-public class BusConsumerTest extends BusTopicTestBase {
+public class BusConsumerTest extends TopicTestBase {
 
     @Before
     public void setUp() {
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java
index 4e78b676..1f99315e 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusPublisherTest.java
@@ -39,13 +39,13 @@ import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher.CambriaPublisherWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher.DmaapAafPublisherWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher.DmaapDmePublisherWrapper;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusPublisher.DmaapPublisherWrapper;
 
-public class BusPublisherTest extends BusTopicTestBase {
+public class BusPublisherTest extends TopicTestBase {
 
     @Before
     public void setUp() {
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBaseTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBaseTest.java
index 01e2e61e..56d3b47a 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBaseTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicBaseTest.java
@@ -27,9 +27,9 @@ import static org.junit.Assert.assertTrue;
 
 import org.junit.Before;
 import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class BusTopicBaseTest extends BusTopicTestBase {
+public class BusTopicBaseTest extends TopicTestBase {
 
     private BusTopicBaseImpl base;
 
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
index d56374fc..8f1ab9bc 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusTopicParamsTest.java
@@ -29,10 +29,10 @@ import java.util.LinkedList;
 import java.util.function.BiConsumer;
 import org.junit.Before;
 import org.junit.Test;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams.TopicParamsBuilder;
 
-public class BusTopicParamsTest extends BusTopicTestBase {
+public class BusTopicParamsTest extends TopicTestBase {
 
     @Before
     public void setUp() {
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
index 69ad43d9..7330e6f7 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineBusTopicSinkTest.java
@@ -36,9 +36,9 @@ import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicListener;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class InlineBusTopicSinkTest extends BusTopicTestBase {
+public class InlineBusTopicSinkTest extends TopicTestBase {
 
     private InlineBusTopicSinkImpl sink;
 
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSinkTest.java
index 2f02a7b7..bf064100 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineDmaapTopicSinkTest.java
@@ -27,9 +27,9 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class InlineDmaapTopicSinkTest extends BusTopicTestBase {
+public class InlineDmaapTopicSinkTest extends TopicTestBase {
     private InlineDmaapTopicSink sink;
 
     /**
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java
index 02714d7c..610070e3 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/InlineUebTopicSinkTest.java
@@ -27,9 +27,9 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class InlineUebTopicSinkTest extends BusTopicTestBase {
+public class InlineUebTopicSinkTest extends TopicTestBase {
     private InlineUebTopicSink sink;
 
     /**
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSourceTest.java
index eaeafef6..d0e8569b 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSourceTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSourceTest.java
@@ -41,10 +41,10 @@ import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicListener;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FilterableBusConsumer;
 
-public class SingleThreadedBusTopicSourceTest extends BusTopicTestBase {
+public class SingleThreadedBusTopicSourceTest extends TopicTestBase {
     private Thread thread;
     private BusConsumer cons;
     private TopicListener listener;
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSourceTest.java
index fa5a67a8..c7444034 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSourceTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedDmaapTopicSourceTest.java
@@ -28,9 +28,9 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class SingleThreadedDmaapTopicSourceTest extends BusTopicTestBase {
+public class SingleThreadedDmaapTopicSourceTest extends TopicTestBase {
     private SingleThreadedDmaapTopicSource source;
 
     /**
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java
index 373e80a4..b8cb0c13 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedUebTopicSourceTest.java
@@ -27,9 +27,9 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class SingleThreadedUebTopicSourceTest extends BusTopicTestBase {
+public class SingleThreadedUebTopicSourceTest extends TopicTestBase {
     private SingleThreadedUebTopicSource source;
 
     /**
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBaseTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBaseTest.java
index 4634d125..76883b3c 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBaseTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/TopicBaseTest.java
@@ -36,9 +36,9 @@ import org.junit.Before;
 import org.junit.Test;
 import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicListener;
-import org.onap.policy.common.endpoints.event.comm.bus.BusTopicTestBase;
+import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
 
-public class TopicBaseTest extends BusTopicTestBase {
+public class TopicBaseTest extends TopicTestBase {
 
     private TopicBaseImpl base;