Remove policy-endpoints checkstyle suppressions

Split class files so each file only contains one class.

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

8 files changed, 1215 insertions, 1128 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 4c36a39a..a3eb4df6 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,18 +21,9 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
-
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.InlineDmaapTopicSink;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * DMAAP Topic Sink Factory.
@@ -121,296 +112,3 @@ public interface DmaapTopicSinkFactory {
      */
     List<DmaapTopicSink> inventory();
 }
-
-
-/* ------------- implementation ----------------- */
-
-/**
- * Factory of DMAAP Reader Topics indexed by topic name.
- */
-class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
-
-    private static final String MISSING_TOPIC = "A topic must be provided";
-
-    /**
-     * Logger.
-     */
-    private static Logger logger = LoggerFactory.getLogger(IndexedDmaapTopicSinkFactory.class);
-
-    /**
-     * DMAAP Topic Name Index.
-     */
-    protected HashMap<String, DmaapTopicSink> dmaapTopicWriters = new HashMap<>();
-
-    @Override
-    public DmaapTopicSink build(BusTopicParams busTopicParams) {
-
-        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (dmaapTopicWriters.containsKey(busTopicParams.getTopic())) {
-                return dmaapTopicWriters.get(busTopicParams.getTopic());
-            }
-
-            DmaapTopicSink dmaapTopicSink = makeSink(busTopicParams);
-
-            if (busTopicParams.isManaged()) {
-                dmaapTopicWriters.put(busTopicParams.getTopic(), dmaapTopicSink);
-            }
-            return dmaapTopicSink;
-        }
-    }
-
-    @Override
-    public DmaapTopicSink build(List<String> servers, String topic) {
-        return this.build(BusTopicParams.builder()
-                .servers(servers)
-                .topic(topic)
-                .managed(true)
-                .useHttps(false)
-                .allowSelfSignedCerts(false)
-                .build());
-    }
-
-    @Override
-    public List<DmaapTopicSink> build(Properties properties) {
-
-        String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS);
-        if (writeTopics == null || writeTopics.isEmpty()) {
-            logger.info("{}: no topic for DMaaP Sink", this);
-            return new ArrayList<>();
-        }
-
-        List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
-        List<DmaapTopicSink> newDmaapTopicSinks = new ArrayList<>();
-        synchronized (this) {
-            for (String topic : writeTopicList) {
-                if (this.dmaapTopicWriters.containsKey(topic)) {
-                    newDmaapTopicSinks.add(this.dmaapTopicWriters.get(topic));
-                    continue;
-                }
-                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
-
-                List<String> serverList;
-                if (servers != null && !servers.isEmpty()) {
-                    serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
-                } else {
-                    serverList = new ArrayList<>();
-                }
-
-                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
-                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
-
-                final String aafMechId = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX);
-                final String aafPassword = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX);
-
-                final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
-
-                final String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
-
-                /* DME2 Properties */
-
-                final String dme2Environment = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
-
-                final String dme2AftEnvironment = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
-
-                final String dme2Partner = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX);
-
-                final String dme2RouteOffer = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX);
-
-                final String dme2Latitude = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
-                                + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
-
-                final String dme2Longitude = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
-
-                final String dme2EpReadTimeoutMs = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX);
-
-                final String dme2EpConnTimeout = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX);
-
-                final String dme2RoundtripTimeoutMs =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX);
-
-                final String dme2Version = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
-                                + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX);
-
-                final String dme2SubContextPath = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX);
-
-                final String dme2SessionStickinessRequired =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
-                                + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX);
-
-                Map<String, String> dme2AdditionalProps = new HashMap<>();
-
-                if (dme2EpReadTimeoutMs != null && !dme2EpReadTimeoutMs.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_READ_TIMEOUT_PROPERTY, dme2EpReadTimeoutMs);
-                }
-                if (dme2EpConnTimeout != null && !dme2EpConnTimeout.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_EP_CONN_TIMEOUT_PROPERTY, dme2EpConnTimeout);
-                }
-                if (dme2RoundtripTimeoutMs != null && !dme2RoundtripTimeoutMs.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_ROUNDTRIP_TIMEOUT_PROPERTY, dme2RoundtripTimeoutMs);
-                }
-                if (dme2Version != null && !dme2Version.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_VERSION_PROPERTY, dme2Version);
-                }
-                if (dme2RouteOffer != null && !dme2RouteOffer.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
-                }
-                if (dme2SubContextPath != null && !dme2SubContextPath.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_SUBCONTEXT_PATH_PROPERTY, dme2SubContextPath);
-                }
-                if (dme2SessionStickinessRequired != null && !dme2SessionStickinessRequired.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_SESSION_STICKINESS_REQUIRED_PROPERTY, dme2SessionStickinessRequired);
-                }
-
-                if (servers == null || servers.isEmpty()) {
-                    logger.error("{}: no DMaaP servers or DME2 ServiceName provided", this);
-                    continue;
-                }
-
-                boolean managed = true;
-                if (managedString != null && !managedString.isEmpty()) {
-                    managed = Boolean.parseBoolean(managedString);
-                }
-
-                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
-
-                // default is to use HTTP if no https property exists
-                boolean useHttps = false;
-                if (useHttpsString != null && !useHttpsString.isEmpty()) {
-                    useHttps = Boolean.parseBoolean(useHttpsString);
-                }
-
-                String allowSelfSignedCertsString =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
-
-                // default is to disallow self-signed certs
-                boolean allowSelfSignedCerts = false;
-                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
-                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
-                }
-
-                DmaapTopicSink dmaapTopicSink = this.build(BusTopicParams.builder()
-                        .servers(serverList)
-                        .topic(topic)
-                        .apiKey(apiKey)
-                        .apiSecret(apiSecret)
-                        .userName(aafMechId)
-                        .password(aafPassword)
-                        .partitionId(partitionKey)
-                        .environment(dme2Environment)
-                        .aftEnvironment(dme2AftEnvironment)
-                        .partner(dme2Partner)
-                        .latitude(dme2Latitude)
-                        .longitude(dme2Longitude)
-                        .additionalProps(dme2AdditionalProps)
-                        .managed(managed)
-                        .useHttps(useHttps)
-                        .allowSelfSignedCerts(allowSelfSignedCerts)
-                        .build());
-
-                newDmaapTopicSinks.add(dmaapTopicSink);
-            }
-            return newDmaapTopicSinks;
-        }
-    }
-
-    /**
-     * Makes a new sink.
-     *
-     * @param busTopicParams parameters to use to configure the sink
-     * @return a new sink
-     */
-    protected DmaapTopicSink makeSink(BusTopicParams busTopicParams) {
-        return new InlineDmaapTopicSink(busTopicParams);
-    }
-
-    @Override
-    public void destroy(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        DmaapTopicSink dmaapTopicWriter;
-        synchronized (this) {
-            if (!dmaapTopicWriters.containsKey(topic)) {
-                return;
-            }
-
-            dmaapTopicWriter = dmaapTopicWriters.remove(topic);
-        }
-
-        dmaapTopicWriter.shutdown();
-    }
-
-    @Override
-    public void destroy() {
-        List<DmaapTopicSink> writers = this.inventory();
-        for (DmaapTopicSink writer : writers) {
-            writer.shutdown();
-        }
-
-        synchronized (this) {
-            this.dmaapTopicWriters.clear();
-        }
-    }
-
-    @Override
-    public DmaapTopicSink get(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (dmaapTopicWriters.containsKey(topic)) {
-                return dmaapTopicWriters.get(topic);
-            } else {
-                throw new IllegalStateException("DmaapTopicSink for " + topic + " not found");
-            }
-        }
-    }
-
-    @Override
-    public synchronized List<DmaapTopicSink> inventory() {
-        return new ArrayList<>(this.dmaapTopicWriters.values());
-    }
-
-    @Override
-    public String toString() {
-        return "IndexedDmaapTopicSinkFactory []";
-    }
-
-}
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 ae6c6c3b..35a79bf1 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,18 +21,9 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
-
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedDmaapTopicSource;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * DMAAP Topic Source Factory.
@@ -119,349 +110,3 @@ public interface DmaapTopicSourceFactory {
      */
     List<DmaapTopicSource> inventory();
 }
-
-
-/* ------------- implementation ----------------- */
-
-/**
- * Factory of DMAAP Source Topics indexed by topic name.
- */
-
-class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
-    private static final String MISSING_TOPIC = "A topic must be provided";
-
-    /**
-     * Logger.
-     */
-    private static Logger logger = LoggerFactory.getLogger(IndexedDmaapTopicSourceFactory.class);
-
-    /**
-     * DMaaP Topic Name Index.
-     */
-    protected HashMap<String, DmaapTopicSource> dmaapTopicSources = new HashMap<>();
-
-    @Override
-    public DmaapTopicSource build(BusTopicParams busTopicParams) {
-
-        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (dmaapTopicSources.containsKey(busTopicParams.getTopic())) {
-                return dmaapTopicSources.get(busTopicParams.getTopic());
-            }
-
-            DmaapTopicSource dmaapTopicSource = makeSource(busTopicParams);
-
-            if (busTopicParams.isManaged()) {
-                dmaapTopicSources.put(busTopicParams.getTopic(), dmaapTopicSource);
-            }
-            return dmaapTopicSource;
-        }
-    }
-
-    @Override
-    public List<DmaapTopicSource> build(Properties properties) {
-
-        String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS);
-        if (readTopics == null || readTopics.isEmpty()) {
-            logger.info("{}: no topic for DMaaP Source", this);
-            return new ArrayList<>();
-        }
-        List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
-
-        List<DmaapTopicSource> dmaapTopicSourceLst = new ArrayList<>();
-        synchronized (this) {
-            for (String topic : readTopicList) {
-                if (this.dmaapTopicSources.containsKey(topic)) {
-                    dmaapTopicSourceLst.add(this.dmaapTopicSources.get(topic));
-                    continue;
-                }
-
-                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
-
-                List<String> serverList;
-                if (servers != null && !servers.isEmpty()) {
-                    serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
-                } else {
-                    serverList = new ArrayList<>();
-                }
-
-                final String apiKey = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
-
-                final String apiSecret = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
-
-                final String aafMechId = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX);
-
-                final String aafPassword = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX);
-
-                final String consumerGroup = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
-
-                final String consumerInstance = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
-
-                final String fetchTimeoutString = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
-
-                /* DME2 Properties */
-
-                final String dme2Environment = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
-
-                final String dme2AftEnvironment = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
-
-                final String dme2Partner = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX);
-
-                final String dme2RouteOffer = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX);
-
-                final String dme2Latitude = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
-
-                final String dme2Longitude = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
-
-                final String dme2EpReadTimeoutMs =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX);
-
-                final String dme2EpConnTimeout = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX);
-
-                final String dme2RoundtripTimeoutMs =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX);
-
-                final String dme2Version = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX);
-
-                final String dme2SubContextPath = properties.getProperty(
-                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX);
-
-                final String dme2SessionStickinessRequired =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX);
-
-                Map<String, String> dme2AdditionalProps = new HashMap<>();
-
-                if (dme2EpReadTimeoutMs != null && !dme2EpReadTimeoutMs.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_READ_TIMEOUT_PROPERTY, dme2EpReadTimeoutMs);
-                }
-                if (dme2EpConnTimeout != null && !dme2EpConnTimeout.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_EP_CONN_TIMEOUT_PROPERTY, dme2EpConnTimeout);
-                }
-                if (dme2RoundtripTimeoutMs != null && !dme2RoundtripTimeoutMs.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_ROUNDTRIP_TIMEOUT_PROPERTY, dme2RoundtripTimeoutMs);
-                }
-                if (dme2Version != null && !dme2Version.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_VERSION_PROPERTY, dme2Version);
-                }
-                if (dme2RouteOffer != null && !dme2RouteOffer.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
-                }
-                if (dme2SubContextPath != null && !dme2SubContextPath.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_SUBCONTEXT_PATH_PROPERTY, dme2SubContextPath);
-                }
-                if (dme2SessionStickinessRequired != null && !dme2SessionStickinessRequired.isEmpty()) {
-                    dme2AdditionalProps.put(DME2_SESSION_STICKINESS_REQUIRED_PROPERTY, dme2SessionStickinessRequired);
-                }
-
-
-                if (servers == null || servers.isEmpty()) {
-
-                    logger.error("{}: no DMaaP servers or DME2 ServiceName provided", this);
-                    continue;
-                }
-
-                int fetchTimeout = DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
-                if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
-                    try {
-                        fetchTimeout = Integer.parseInt(fetchTimeoutString);
-                    } catch (NumberFormatException nfe) {
-                        logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
-                                topic);
-                    }
-                }
-
-                String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
-                int fetchLimit = DmaapTopicSource.DEFAULT_LIMIT_FETCH;
-                if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
-                    try {
-                        fetchLimit = Integer.parseInt(fetchLimitString);
-                    } catch (NumberFormatException nfe) {
-                        logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
-                                topic);
-                    }
-                }
-
-                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
-                boolean managed = true;
-                if (managedString != null && !managedString.isEmpty()) {
-                    managed = Boolean.parseBoolean(managedString);
-                }
-
-                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
-
-                // default is to use HTTP if no https property exists
-                boolean useHttps = false;
-                if (useHttpsString != null && !useHttpsString.isEmpty()) {
-                    useHttps = Boolean.parseBoolean(useHttpsString);
-                }
-
-                String allowSelfSignedCertsString =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
-
-                // default is to disallow self-signed certs
-                boolean allowSelfSignedCerts = false;
-                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
-                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
-                }
-
-
-                DmaapTopicSource uebTopicSource = this.build(BusTopicParams.builder()
-                        .servers(serverList)
-                        .topic(topic)
-                        .apiKey(apiKey)
-                        .apiSecret(apiSecret)
-                        .userName(aafMechId)
-                        .password(aafPassword)
-                        .consumerGroup(consumerGroup)
-                        .consumerInstance(consumerInstance)
-                        .fetchTimeout(fetchTimeout)
-                        .fetchLimit(fetchLimit)
-                        .environment(dme2Environment)
-                        .aftEnvironment(dme2AftEnvironment)
-                        .partner(dme2Partner)
-                        .latitude(dme2Latitude)
-                        .longitude(dme2Longitude)
-                        .additionalProps(dme2AdditionalProps)
-                        .managed(managed)
-                        .useHttps(useHttps)
-                        .allowSelfSignedCerts(allowSelfSignedCerts)
-                        .build());
-
-                dmaapTopicSourceLst.add(uebTopicSource);
-            }
-        }
-        return dmaapTopicSourceLst;
-    }
-
-    @Override
-    public DmaapTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
-        return this.build(BusTopicParams.builder()
-                .servers(servers)
-                .topic(topic)
-                .apiKey(apiKey)
-                .apiSecret(apiSecret)
-                .fetchTimeout(DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH)
-                .fetchLimit(DmaapTopicSource.DEFAULT_LIMIT_FETCH)
-                .managed(true)
-                .useHttps(false)
-                .allowSelfSignedCerts(false)
-                .build());
-    }
-
-    @Override
-    public DmaapTopicSource build(List<String> servers, String topic) {
-        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 DmaapTopicSource makeSource(BusTopicParams busTopicParams) {
-        return new SingleThreadedDmaapTopicSource(busTopicParams);
-    }
-
-    @Override
-    public void destroy(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        DmaapTopicSource uebTopicSource;
-
-        synchronized (this) {
-            if (!dmaapTopicSources.containsKey(topic)) {
-                return;
-            }
-
-            uebTopicSource = dmaapTopicSources.remove(topic);
-        }
-
-        uebTopicSource.shutdown();
-    }
-
-    @Override
-    public void destroy() {
-        List<DmaapTopicSource> readers = this.inventory();
-        for (DmaapTopicSource reader : readers) {
-            reader.shutdown();
-        }
-
-        synchronized (this) {
-            this.dmaapTopicSources.clear();
-        }
-    }
-
-    @Override
-    public DmaapTopicSource get(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (dmaapTopicSources.containsKey(topic)) {
-                return dmaapTopicSources.get(topic);
-            } else {
-                throw new IllegalStateException("DmaapTopiceSource for " + topic + " not found");
-            }
-        }
-    }
-
-    @Override
-    public synchronized List<DmaapTopicSource> inventory() {
-        return new ArrayList<>(this.dmaapTopicSources.values());
-    }
-
-    @Override
-    public String toString() {
-        return "IndexedDmaapTopicSourceFactory []";
-    }
-
-}
-
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSinkFactory.java
new file mode 100644
index 00000000..659833ce
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSinkFactory.java
@@ -0,0 +1,324 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * Copyright (C) 2017-2019 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.InlineDmaapTopicSink;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory of DMAAP Reader Topics indexed by topic name.
+ */
+class IndexedDmaapTopicSinkFactory implements DmaapTopicSinkFactory {
+
+    private static final String MISSING_TOPIC = "A topic must be provided";
+
+    /**
+     * Logger.
+     */
+    private static Logger logger = LoggerFactory.getLogger(IndexedDmaapTopicSinkFactory.class);
+
+    /**
+     * DMAAP Topic Name Index.
+     */
+    protected HashMap<String, DmaapTopicSink> dmaapTopicWriters = new HashMap<>();
+
+    @Override
+    public DmaapTopicSink build(BusTopicParams busTopicParams) {
+
+        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (dmaapTopicWriters.containsKey(busTopicParams.getTopic())) {
+                return dmaapTopicWriters.get(busTopicParams.getTopic());
+            }
+
+            DmaapTopicSink dmaapTopicSink = makeSink(busTopicParams);
+
+            if (busTopicParams.isManaged()) {
+                dmaapTopicWriters.put(busTopicParams.getTopic(), dmaapTopicSink);
+            }
+            return dmaapTopicSink;
+        }
+    }
+
+    @Override
+    public DmaapTopicSink build(List<String> servers, String topic) {
+        return this.build(BusTopicParams.builder()
+                .servers(servers)
+                .topic(topic)
+                .managed(true)
+                .useHttps(false)
+                .allowSelfSignedCerts(false)
+                .build());
+    }
+
+    @Override
+    public List<DmaapTopicSink> build(Properties properties) {
+
+        String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS);
+        if (writeTopics == null || writeTopics.isEmpty()) {
+            logger.info("{}: no topic for DMaaP Sink", this);
+            return new ArrayList<>();
+        }
+
+        List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
+        List<DmaapTopicSink> newDmaapTopicSinks = new ArrayList<>();
+        synchronized (this) {
+            for (String topic : writeTopicList) {
+                if (this.dmaapTopicWriters.containsKey(topic)) {
+                    newDmaapTopicSinks.add(this.dmaapTopicWriters.get(topic));
+                    continue;
+                }
+                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
+
+                List<String> serverList;
+                if (servers != null && !servers.isEmpty()) {
+                    serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
+                } else {
+                    serverList = new ArrayList<>();
+                }
+
+                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
+                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
+
+                final String aafMechId = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX);
+                final String aafPassword = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX);
+
+                final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
+
+                final String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
+
+                /* DME2 Properties */
+
+                final String dme2Environment = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
+
+                final String dme2AftEnvironment = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
+
+                final String dme2Partner = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX);
+
+                final String dme2RouteOffer = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX);
+
+                final String dme2Latitude = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
+                                + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
+
+                final String dme2Longitude = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
+
+                final String dme2EpReadTimeoutMs = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX);
+
+                final String dme2EpConnTimeout = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX);
+
+                final String dme2RoundtripTimeoutMs =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX);
+
+                final String dme2Version = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
+                                + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX);
+
+                final String dme2SubContextPath = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX);
+
+                final String dme2SessionStickinessRequired =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS
+                                + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX);
+
+                Map<String, String> dme2AdditionalProps = new HashMap<>();
+
+                if (dme2EpReadTimeoutMs != null && !dme2EpReadTimeoutMs.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_READ_TIMEOUT_PROPERTY, dme2EpReadTimeoutMs);
+                }
+                if (dme2EpConnTimeout != null && !dme2EpConnTimeout.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_EP_CONN_TIMEOUT_PROPERTY, dme2EpConnTimeout);
+                }
+                if (dme2RoundtripTimeoutMs != null && !dme2RoundtripTimeoutMs.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_ROUNDTRIP_TIMEOUT_PROPERTY, dme2RoundtripTimeoutMs);
+                }
+                if (dme2Version != null && !dme2Version.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_VERSION_PROPERTY, dme2Version);
+                }
+                if (dme2RouteOffer != null && !dme2RouteOffer.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
+                }
+                if (dme2SubContextPath != null && !dme2SubContextPath.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_SUBCONTEXT_PATH_PROPERTY, dme2SubContextPath);
+                }
+                if (dme2SessionStickinessRequired != null && !dme2SessionStickinessRequired.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_SESSION_STICKINESS_REQUIRED_PROPERTY, dme2SessionStickinessRequired);
+                }
+
+                if (servers == null || servers.isEmpty()) {
+                    logger.error("{}: no DMaaP servers or DME2 ServiceName provided", this);
+                    continue;
+                }
+
+                boolean managed = true;
+                if (managedString != null && !managedString.isEmpty()) {
+                    managed = Boolean.parseBoolean(managedString);
+                }
+
+                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
+
+                // default is to use HTTP if no https property exists
+                boolean useHttps = false;
+                if (useHttpsString != null && !useHttpsString.isEmpty()) {
+                    useHttps = Boolean.parseBoolean(useHttpsString);
+                }
+
+                String allowSelfSignedCertsString =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
+
+                // default is to disallow self-signed certs
+                boolean allowSelfSignedCerts = false;
+                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
+                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
+                }
+
+                DmaapTopicSink dmaapTopicSink = this.build(BusTopicParams.builder()
+                        .servers(serverList)
+                        .topic(topic)
+                        .apiKey(apiKey)
+                        .apiSecret(apiSecret)
+                        .userName(aafMechId)
+                        .password(aafPassword)
+                        .partitionId(partitionKey)
+                        .environment(dme2Environment)
+                        .aftEnvironment(dme2AftEnvironment)
+                        .partner(dme2Partner)
+                        .latitude(dme2Latitude)
+                        .longitude(dme2Longitude)
+                        .additionalProps(dme2AdditionalProps)
+                        .managed(managed)
+                        .useHttps(useHttps)
+                        .allowSelfSignedCerts(allowSelfSignedCerts)
+                        .build());
+
+                newDmaapTopicSinks.add(dmaapTopicSink);
+            }
+            return newDmaapTopicSinks;
+        }
+    }
+
+    /**
+     * Makes a new sink.
+     *
+     * @param busTopicParams parameters to use to configure the sink
+     * @return a new sink
+     */
+    protected DmaapTopicSink makeSink(BusTopicParams busTopicParams) {
+        return new InlineDmaapTopicSink(busTopicParams);
+    }
+
+    @Override
+    public void destroy(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        DmaapTopicSink dmaapTopicWriter;
+        synchronized (this) {
+            if (!dmaapTopicWriters.containsKey(topic)) {
+                return;
+            }
+
+            dmaapTopicWriter = dmaapTopicWriters.remove(topic);
+        }
+
+        dmaapTopicWriter.shutdown();
+    }
+
+    @Override
+    public void destroy() {
+        List<DmaapTopicSink> writers = this.inventory();
+        for (DmaapTopicSink writer : writers) {
+            writer.shutdown();
+        }
+
+        synchronized (this) {
+            this.dmaapTopicWriters.clear();
+        }
+    }
+
+    @Override
+    public DmaapTopicSink get(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (dmaapTopicWriters.containsKey(topic)) {
+                return dmaapTopicWriters.get(topic);
+            } else {
+                throw new IllegalStateException("DmaapTopicSink for " + topic + " not found");
+            }
+        }
+    }
+
+    @Override
+    public synchronized List<DmaapTopicSink> inventory() {
+        return new ArrayList<>(this.dmaapTopicWriters.values());
+    }
+
+    @Override
+    public String toString() {
+        return "IndexedDmaapTopicSinkFactory []";
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSourceFactory.java
new file mode 100644
index 00000000..0c008f11
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedDmaapTopicSourceFactory.java
@@ -0,0 +1,376 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * Copyright (C) 2017-2019 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedDmaapTopicSource;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory of DMAAP Source Topics indexed by topic name.
+ */
+
+class IndexedDmaapTopicSourceFactory implements DmaapTopicSourceFactory {
+    private static final String MISSING_TOPIC = "A topic must be provided";
+
+    /**
+     * Logger.
+     */
+    private static Logger logger = LoggerFactory.getLogger(IndexedDmaapTopicSourceFactory.class);
+
+    /**
+     * DMaaP Topic Name Index.
+     */
+    protected HashMap<String, DmaapTopicSource> dmaapTopicSources = new HashMap<>();
+
+    @Override
+    public DmaapTopicSource build(BusTopicParams busTopicParams) {
+
+        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (dmaapTopicSources.containsKey(busTopicParams.getTopic())) {
+                return dmaapTopicSources.get(busTopicParams.getTopic());
+            }
+
+            DmaapTopicSource dmaapTopicSource = makeSource(busTopicParams);
+
+            if (busTopicParams.isManaged()) {
+                dmaapTopicSources.put(busTopicParams.getTopic(), dmaapTopicSource);
+            }
+            return dmaapTopicSource;
+        }
+    }
+
+    @Override
+    public List<DmaapTopicSource> build(Properties properties) {
+
+        String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS);
+        if (readTopics == null || readTopics.isEmpty()) {
+            logger.info("{}: no topic for DMaaP Source", this);
+            return new ArrayList<>();
+        }
+        List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
+
+        List<DmaapTopicSource> dmaapTopicSourceLst = new ArrayList<>();
+        synchronized (this) {
+            for (String topic : readTopicList) {
+                if (this.dmaapTopicSources.containsKey(topic)) {
+                    dmaapTopicSourceLst.add(this.dmaapTopicSources.get(topic));
+                    continue;
+                }
+
+                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
+
+                List<String> serverList;
+                if (servers != null && !servers.isEmpty()) {
+                    serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
+                } else {
+                    serverList = new ArrayList<>();
+                }
+
+                final String apiKey = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
+
+                final String apiSecret = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
+
+                final String aafMechId = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_MECHID_SUFFIX);
+
+                final String aafPassword = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX);
+
+                final String consumerGroup = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
+
+                final String consumerInstance = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
+
+                final String fetchTimeoutString = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
+
+                /* DME2 Properties */
+
+                final String dme2Environment = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ENVIRONMENT_SUFFIX);
+
+                final String dme2AftEnvironment = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_AFT_ENVIRONMENT_SUFFIX);
+
+                final String dme2Partner = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_PARTNER_SUFFIX);
+
+                final String dme2RouteOffer = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUTE_OFFER_SUFFIX);
+
+                final String dme2Latitude = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LATITUDE_SUFFIX);
+
+                final String dme2Longitude = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_LONGITUDE_SUFFIX);
+
+                final String dme2EpReadTimeoutMs =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_READ_TIMEOUT_MS_SUFFIX);
+
+                final String dme2EpConnTimeout = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_EP_CONN_TIMEOUT_SUFFIX);
+
+                final String dme2RoundtripTimeoutMs =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_ROUNDTRIP_TIMEOUT_MS_SUFFIX);
+
+                final String dme2Version = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_VERSION_SUFFIX);
+
+                final String dme2SubContextPath = properties.getProperty(
+                                PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SUB_CONTEXT_PATH_SUFFIX);
+
+                final String dme2SessionStickinessRequired =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_DMAAP_DME2_SESSION_STICKINESS_REQUIRED_SUFFIX);
+
+                Map<String, String> dme2AdditionalProps = new HashMap<>();
+
+                if (dme2EpReadTimeoutMs != null && !dme2EpReadTimeoutMs.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_READ_TIMEOUT_PROPERTY, dme2EpReadTimeoutMs);
+                }
+                if (dme2EpConnTimeout != null && !dme2EpConnTimeout.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_EP_CONN_TIMEOUT_PROPERTY, dme2EpConnTimeout);
+                }
+                if (dme2RoundtripTimeoutMs != null && !dme2RoundtripTimeoutMs.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_ROUNDTRIP_TIMEOUT_PROPERTY, dme2RoundtripTimeoutMs);
+                }
+                if (dme2Version != null && !dme2Version.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_VERSION_PROPERTY, dme2Version);
+                }
+                if (dme2RouteOffer != null && !dme2RouteOffer.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_ROUTE_OFFER_PROPERTY, dme2RouteOffer);
+                }
+                if (dme2SubContextPath != null && !dme2SubContextPath.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_SUBCONTEXT_PATH_PROPERTY, dme2SubContextPath);
+                }
+                if (dme2SessionStickinessRequired != null && !dme2SessionStickinessRequired.isEmpty()) {
+                    dme2AdditionalProps.put(DME2_SESSION_STICKINESS_REQUIRED_PROPERTY, dme2SessionStickinessRequired);
+                }
+
+
+                if (servers == null || servers.isEmpty()) {
+
+                    logger.error("{}: no DMaaP servers or DME2 ServiceName provided", this);
+                    continue;
+                }
+
+                int fetchTimeout = DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
+                if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
+                    try {
+                        fetchTimeout = Integer.parseInt(fetchTimeoutString);
+                    } catch (NumberFormatException nfe) {
+                        logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
+                                topic);
+                    }
+                }
+
+                String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
+                int fetchLimit = DmaapTopicSource.DEFAULT_LIMIT_FETCH;
+                if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
+                    try {
+                        fetchLimit = Integer.parseInt(fetchLimitString);
+                    } catch (NumberFormatException nfe) {
+                        logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
+                                topic);
+                    }
+                }
+
+                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
+                boolean managed = true;
+                if (managedString != null && !managedString.isEmpty()) {
+                    managed = Boolean.parseBoolean(managedString);
+                }
+
+                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
+
+                // default is to use HTTP if no https property exists
+                boolean useHttps = false;
+                if (useHttpsString != null && !useHttpsString.isEmpty()) {
+                    useHttps = Boolean.parseBoolean(useHttpsString);
+                }
+
+                String allowSelfSignedCertsString =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_DMAAP_SOURCE_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
+
+                // default is to disallow self-signed certs
+                boolean allowSelfSignedCerts = false;
+                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
+                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
+                }
+
+
+                DmaapTopicSource uebTopicSource = this.build(BusTopicParams.builder()
+                        .servers(serverList)
+                        .topic(topic)
+                        .apiKey(apiKey)
+                        .apiSecret(apiSecret)
+                        .userName(aafMechId)
+                        .password(aafPassword)
+                        .consumerGroup(consumerGroup)
+                        .consumerInstance(consumerInstance)
+                        .fetchTimeout(fetchTimeout)
+                        .fetchLimit(fetchLimit)
+                        .environment(dme2Environment)
+                        .aftEnvironment(dme2AftEnvironment)
+                        .partner(dme2Partner)
+                        .latitude(dme2Latitude)
+                        .longitude(dme2Longitude)
+                        .additionalProps(dme2AdditionalProps)
+                        .managed(managed)
+                        .useHttps(useHttps)
+                        .allowSelfSignedCerts(allowSelfSignedCerts)
+                        .build());
+
+                dmaapTopicSourceLst.add(uebTopicSource);
+            }
+        }
+        return dmaapTopicSourceLst;
+    }
+
+    @Override
+    public DmaapTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
+        return this.build(BusTopicParams.builder()
+                .servers(servers)
+                .topic(topic)
+                .apiKey(apiKey)
+                .apiSecret(apiSecret)
+                .fetchTimeout(DmaapTopicSource.DEFAULT_TIMEOUT_MS_FETCH)
+                .fetchLimit(DmaapTopicSource.DEFAULT_LIMIT_FETCH)
+                .managed(true)
+                .useHttps(false)
+                .allowSelfSignedCerts(false)
+                .build());
+    }
+
+    @Override
+    public DmaapTopicSource build(List<String> servers, String topic) {
+        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 DmaapTopicSource makeSource(BusTopicParams busTopicParams) {
+        return new SingleThreadedDmaapTopicSource(busTopicParams);
+    }
+
+    @Override
+    public void destroy(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        DmaapTopicSource uebTopicSource;
+
+        synchronized (this) {
+            if (!dmaapTopicSources.containsKey(topic)) {
+                return;
+            }
+
+            uebTopicSource = dmaapTopicSources.remove(topic);
+        }
+
+        uebTopicSource.shutdown();
+    }
+
+    @Override
+    public void destroy() {
+        List<DmaapTopicSource> readers = this.inventory();
+        for (DmaapTopicSource reader : readers) {
+            reader.shutdown();
+        }
+
+        synchronized (this) {
+            this.dmaapTopicSources.clear();
+        }
+    }
+
+    @Override
+    public DmaapTopicSource get(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (dmaapTopicSources.containsKey(topic)) {
+                return dmaapTopicSources.get(topic);
+            } else {
+                throw new IllegalStateException("DmaapTopiceSource for " + topic + " not found");
+            }
+        }
+    }
+
+    @Override
+    public synchronized List<DmaapTopicSource> inventory() {
+        return new ArrayList<>(this.dmaapTopicSources.values());
+    }
+
+    @Override
+    public String toString() {
+        return "IndexedDmaapTopicSourceFactory []";
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSinkFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSinkFactory.java
new file mode 100644
index 00000000..5b3fc669
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSinkFactory.java
@@ -0,0 +1,237 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * Copyright (C) 2017-2019 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.InlineUebTopicSink;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory of UEB Reader Topics indexed by topic name.
+ */
+class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
+    private static final String MISSING_TOPIC = "A topic must be provided";
+
+    /**
+     * Logger.
+     */
+    private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
+
+    /**
+     * UEB Topic Name Index.
+     */
+    protected HashMap<String, UebTopicSink> uebTopicSinks = new HashMap<>();
+
+    @Override
+    public UebTopicSink build(BusTopicParams busTopicParams) {
+
+        if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
+            throw new IllegalArgumentException("UEB Server(s) must be provided");
+        }
+
+        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (uebTopicSinks.containsKey(busTopicParams.getTopic())) {
+                return uebTopicSinks.get(busTopicParams.getTopic());
+            }
+
+            UebTopicSink uebTopicWriter = makeSink(busTopicParams);
+
+            if (busTopicParams.isManaged()) {
+                uebTopicSinks.put(busTopicParams.getTopic(), uebTopicWriter);
+            }
+
+            return uebTopicWriter;
+        }
+    }
+
+
+    @Override
+    public UebTopicSink build(List<String> servers, String topic) {
+        return this.build(BusTopicParams.builder()
+                .servers(servers)
+                .topic(topic)
+                .managed(true)
+                .useHttps(false)
+                .allowSelfSignedCerts(false)
+                .build());
+    }
+
+
+    @Override
+    public List<UebTopicSink> build(Properties properties) {
+
+        String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS);
+        if (writeTopics == null || writeTopics.isEmpty()) {
+            logger.info("{}: no topic for UEB Sink", this);
+            return new ArrayList<>();
+        }
+
+        List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
+        List<UebTopicSink> newUebTopicSinks = new ArrayList<>();
+        synchronized (this) {
+            for (String topic : writeTopicList) {
+                if (this.uebTopicSinks.containsKey(topic)) {
+                    newUebTopicSinks.add(this.uebTopicSinks.get(topic));
+                    continue;
+                }
+
+                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
+                        + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
+                if (servers == null || servers.isEmpty()) {
+                    logger.error("{}: no UEB servers configured for sink {}", this, topic);
+                    continue;
+                }
+
+                final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
+
+                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
+                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
+                final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
+                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
+
+                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
+                boolean managed = true;
+                if (managedString != null && !managedString.isEmpty()) {
+                    managed = Boolean.parseBoolean(managedString);
+                }
+
+                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
+
+                // default is to use HTTP if no https property exists
+                boolean useHttps = false;
+                if (useHttpsString != null && !useHttpsString.isEmpty()) {
+                    useHttps = Boolean.parseBoolean(useHttpsString);
+                }
+
+
+                String allowSelfSignedCertsString =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
+
+                // default is to disallow self-signed certs
+                boolean allowSelfSignedCerts = false;
+                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
+                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
+                }
+
+                UebTopicSink uebTopicWriter = this.build(BusTopicParams.builder()
+                        .servers(serverList)
+                        .topic(topic)
+                        .apiKey(apiKey)
+                        .apiSecret(apiSecret)
+                        .partitionId(partitionKey)
+                        .managed(managed)
+                        .useHttps(useHttps)
+                        .allowSelfSignedCerts(allowSelfSignedCerts)
+                        .build());
+                newUebTopicSinks.add(uebTopicWriter);
+            }
+            return newUebTopicSinks;
+        }
+    }
+
+    @Override
+    public void destroy(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        UebTopicSink uebTopicWriter;
+        synchronized (this) {
+            if (!uebTopicSinks.containsKey(topic)) {
+                return;
+            }
+
+            uebTopicWriter = uebTopicSinks.remove(topic);
+        }
+
+        uebTopicWriter.shutdown();
+    }
+
+    @Override
+    public void destroy() {
+        List<UebTopicSink> writers = this.inventory();
+        for (UebTopicSink writer : writers) {
+            writer.shutdown();
+        }
+
+        synchronized (this) {
+            this.uebTopicSinks.clear();
+        }
+    }
+
+    @Override
+    public UebTopicSink get(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (uebTopicSinks.containsKey(topic)) {
+                return uebTopicSinks.get(topic);
+            } else {
+                throw new IllegalStateException("UebTopicSink for " + topic + " not found");
+            }
+        }
+    }
+
+    @Override
+    public synchronized List<UebTopicSink> inventory() {
+        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() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("IndexedUebTopicSinkFactory []");
+        return builder.toString();
+    }
+
+}
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSourceFactory.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSourceFactory.java
new file mode 100644
index 00000000..88a472c2
--- /dev/null
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/IndexedUebTopicSourceFactory.java
@@ -0,0 +1,274 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP Policy Engine - Common Modules
+ * ================================================================================
+ * Copyright (C) 2017-2019 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 java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
+
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedUebTopicSource;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory of UEB Source Topics indexed by topic name.
+ */
+class IndexedUebTopicSourceFactory implements UebTopicSourceFactory {
+    private static final String MISSING_TOPIC = "A topic must be provided";
+
+    /**
+     * Logger.
+     */
+    private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSourceFactory.class);
+
+    /**
+     * UEB Topic Name Index.
+     */
+    protected HashMap<String, UebTopicSource> uebTopicSources = new HashMap<>();
+
+    @Override
+    public UebTopicSource build(BusTopicParams busTopicParams) {
+        if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
+            throw new IllegalArgumentException("UEB Server(s) must be provided");
+        }
+
+        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (uebTopicSources.containsKey(busTopicParams.getTopic())) {
+                return uebTopicSources.get(busTopicParams.getTopic());
+            }
+
+            UebTopicSource uebTopicSource = makeSource(busTopicParams);
+
+            if (busTopicParams.isManaged()) {
+                uebTopicSources.put(busTopicParams.getTopic(), uebTopicSource);
+            }
+
+            return uebTopicSource;
+        }
+    }
+
+    @Override
+    public List<UebTopicSource> build(Properties properties) {
+
+        String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS);
+        if (readTopics == null || readTopics.isEmpty()) {
+            logger.info("{}: no topic for UEB Source", this);
+            return new ArrayList<>();
+        }
+        List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
+
+        List<UebTopicSource> newUebTopicSources = new ArrayList<>();
+        synchronized (this) {
+            for (String topic : readTopicList) {
+                if (this.uebTopicSources.containsKey(topic)) {
+                    newUebTopicSources.add(this.uebTopicSources.get(topic));
+                    continue;
+                }
+
+                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
+
+                if (servers == null || servers.isEmpty()) {
+                    logger.error("{}: no UEB servers configured for sink {}", this, topic);
+                    continue;
+                }
+
+                final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
+
+                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
+
+                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
+
+                final String consumerGroup = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
+
+                final String consumerInstance = properties.getProperty(
+                        PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
+
+                String fetchTimeoutString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
+                int fetchTimeout = UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
+                if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
+                    try {
+                        fetchTimeout = Integer.parseInt(fetchTimeoutString);
+                    } catch (NumberFormatException nfe) {
+                        logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
+                                topic);
+                    }
+                }
+
+                String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
+                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
+                int fetchLimit = UebTopicSource.DEFAULT_LIMIT_FETCH;
+                if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
+                    try {
+                        fetchLimit = Integer.parseInt(fetchLimitString);
+                    } catch (NumberFormatException nfe) {
+                        logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
+                                topic);
+                    }
+                }
+
+                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
+                boolean managed = true;
+                if (managedString != null && !managedString.isEmpty()) {
+                    managed = Boolean.parseBoolean(managedString);
+                }
+
+                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
+                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
+
+                // default is to use HTTP if no https property exists
+                boolean useHttps = false;
+                if (useHttpsString != null && !useHttpsString.isEmpty()) {
+                    useHttps = Boolean.parseBoolean(useHttpsString);
+                }
+
+                String allowSelfSignedCertsString =
+                        properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
+                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
+
+                // default is to disallow self-signed certs
+                boolean allowSelfSignedCerts = false;
+                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
+                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
+                }
+
+                UebTopicSource uebTopicSource = this.build(BusTopicParams.builder()
+                        .servers(serverList)
+                        .topic(topic)
+                        .apiKey(apiKey)
+                        .apiSecret(apiSecret)
+                        .consumerGroup(consumerGroup)
+                        .consumerInstance(consumerInstance)
+                        .fetchTimeout(fetchTimeout)
+                        .fetchLimit(fetchLimit)
+                        .managed(managed)
+                        .useHttps(useHttps)
+                        .allowSelfSignedCerts(allowSelfSignedCerts).build());
+                newUebTopicSources.add(uebTopicSource);
+            }
+        }
+        return newUebTopicSources;
+    }
+
+    @Override
+    public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
+
+        return this.build(BusTopicParams.builder()
+                .servers(servers)
+                .topic(topic)
+                .apiKey(apiKey)
+                .apiSecret(apiSecret)
+                .fetchTimeout(UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH)
+                .fetchLimit(UebTopicSource.DEFAULT_LIMIT_FETCH)
+                .managed(true)
+                .useHttps(false)
+                .allowSelfSignedCerts(true).build());
+    }
+
+    @Override
+    public UebTopicSource build(List<String> servers, String topic) {
+        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) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        UebTopicSource uebTopicSource;
+
+        synchronized (this) {
+            if (!uebTopicSources.containsKey(topic)) {
+                return;
+            }
+
+            uebTopicSource = uebTopicSources.remove(topic);
+        }
+
+        uebTopicSource.shutdown();
+    }
+
+    @Override
+    public void destroy() {
+        List<UebTopicSource> readers = this.inventory();
+        for (UebTopicSource reader : readers) {
+            reader.shutdown();
+        }
+
+        synchronized (this) {
+            this.uebTopicSources.clear();
+        }
+    }
+
+    @Override
+    public UebTopicSource get(String topic) {
+
+        if (topic == null || topic.isEmpty()) {
+            throw new IllegalArgumentException(MISSING_TOPIC);
+        }
+
+        synchronized (this) {
+            if (uebTopicSources.containsKey(topic)) {
+                return uebTopicSources.get(topic);
+            } else {
+                throw new IllegalStateException("UebTopiceSource for " + topic + " not found");
+            }
+        }
+    }
+
+    @Override
+    public synchronized List<UebTopicSource> inventory() {
+        return new ArrayList<>(this.uebTopicSources.values());
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder builder = new StringBuilder();
+        builder.append("IndexedUebTopicSourceFactory []");
+        return builder.toString();
+    }
+}
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 c200af5a..0cf095fd 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,17 +21,9 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Properties;
-
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.InlineUebTopicSink;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * UEB Topic Sink Factory.
@@ -98,210 +90,3 @@ public interface UebTopicSinkFactory {
      */
     List<UebTopicSink> inventory();
 }
-
-
-/* ------------- implementation ----------------- */
-
-/**
- * Factory of UEB Reader Topics indexed by topic name.
- */
-class IndexedUebTopicSinkFactory implements UebTopicSinkFactory {
-    private static final String MISSING_TOPIC = "A topic must be provided";
-
-    /**
-     * Logger.
-     */
-    private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSinkFactory.class);
-
-    /**
-     * UEB Topic Name Index.
-     */
-    protected HashMap<String, UebTopicSink> uebTopicSinks = new HashMap<>();
-
-    @Override
-    public UebTopicSink build(BusTopicParams busTopicParams) {
-
-        if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
-            throw new IllegalArgumentException("UEB Server(s) must be provided");
-        }
-
-        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (uebTopicSinks.containsKey(busTopicParams.getTopic())) {
-                return uebTopicSinks.get(busTopicParams.getTopic());
-            }
-
-            UebTopicSink uebTopicWriter = makeSink(busTopicParams);
-
-            if (busTopicParams.isManaged()) {
-                uebTopicSinks.put(busTopicParams.getTopic(), uebTopicWriter);
-            }
-
-            return uebTopicWriter;
-        }
-    }
-
-
-    @Override
-    public UebTopicSink build(List<String> servers, String topic) {
-        return this.build(BusTopicParams.builder()
-                .servers(servers)
-                .topic(topic)
-                .managed(true)
-                .useHttps(false)
-                .allowSelfSignedCerts(false)
-                .build());
-    }
-
-
-    @Override
-    public List<UebTopicSink> build(Properties properties) {
-
-        String writeTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS);
-        if (writeTopics == null || writeTopics.isEmpty()) {
-            logger.info("{}: no topic for UEB Sink", this);
-            return new ArrayList<>();
-        }
-
-        List<String> writeTopicList = new ArrayList<>(Arrays.asList(writeTopics.split("\\s*,\\s*")));
-        List<UebTopicSink> newUebTopicSinks = new ArrayList<>();
-        synchronized (this) {
-            for (String topic : writeTopicList) {
-                if (this.uebTopicSinks.containsKey(topic)) {
-                    newUebTopicSinks.add(this.uebTopicSinks.get(topic));
-                    continue;
-                }
-
-                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
-                        + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
-                if (servers == null || servers.isEmpty()) {
-                    logger.error("{}: no UEB servers configured for sink {}", this, topic);
-                    continue;
-                }
-
-                final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
-
-                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
-                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
-                final String partitionKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS 
-                                + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SINK_PARTITION_KEY_SUFFIX);
-
-                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
-                boolean managed = true;
-                if (managedString != null && !managedString.isEmpty()) {
-                    managed = Boolean.parseBoolean(managedString);
-                }
-
-                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
-
-                // default is to use HTTP if no https property exists
-                boolean useHttps = false;
-                if (useHttpsString != null && !useHttpsString.isEmpty()) {
-                    useHttps = Boolean.parseBoolean(useHttpsString);
-                }
-
-
-                String allowSelfSignedCertsString =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SINK_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
-
-                // default is to disallow self-signed certs
-                boolean allowSelfSignedCerts = false;
-                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
-                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
-                }
-
-                UebTopicSink uebTopicWriter = this.build(BusTopicParams.builder()
-                        .servers(serverList)
-                        .topic(topic)
-                        .apiKey(apiKey)
-                        .apiSecret(apiSecret)
-                        .partitionId(partitionKey)
-                        .managed(managed)
-                        .useHttps(useHttps)
-                        .allowSelfSignedCerts(allowSelfSignedCerts)
-                        .build());
-                newUebTopicSinks.add(uebTopicWriter);
-            }
-            return newUebTopicSinks;
-        }
-    }
-
-    @Override
-    public void destroy(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        UebTopicSink uebTopicWriter;
-        synchronized (this) {
-            if (!uebTopicSinks.containsKey(topic)) {
-                return;
-            }
-
-            uebTopicWriter = uebTopicSinks.remove(topic);
-        }
-
-        uebTopicWriter.shutdown();
-    }
-
-    @Override
-    public void destroy() {
-        List<UebTopicSink> writers = this.inventory();
-        for (UebTopicSink writer : writers) {
-            writer.shutdown();
-        }
-
-        synchronized (this) {
-            this.uebTopicSinks.clear();
-        }
-    }
-
-    @Override
-    public UebTopicSink get(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (uebTopicSinks.containsKey(topic)) {
-                return uebTopicSinks.get(topic);
-            } else {
-                throw new IllegalStateException("UebTopicSink for " + topic + " not found");
-            }
-        }
-    }
-
-    @Override
-    public synchronized List<UebTopicSink> inventory() {
-        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() {
-        StringBuilder builder = new StringBuilder();
-        builder.append("IndexedUebTopicSinkFactory []");
-        return builder.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 96315881..beacee3b 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
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * Modified Copyright (C) 2018 Samsung Electronics Co., Ltd.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -21,17 +21,9 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Properties;
-
 import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.SingleThreadedUebTopicSource;
-import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * UEB Topic Source Factory.
@@ -110,247 +102,3 @@ public interface UebTopicSourceFactory {
      */
     List<UebTopicSource> inventory();
 }
-
-
-/* ------------- implementation ----------------- */
-
-/**
- * Factory of UEB Source Topics indexed by topic name.
- */
-class IndexedUebTopicSourceFactory implements UebTopicSourceFactory {
-    private static final String MISSING_TOPIC = "A topic must be provided";
-
-    /**
-     * Logger.
-     */
-    private static Logger logger = LoggerFactory.getLogger(IndexedUebTopicSourceFactory.class);
-
-    /**
-     * UEB Topic Name Index.
-     */
-    protected HashMap<String, UebTopicSource> uebTopicSources = new HashMap<>();
-
-    @Override
-    public UebTopicSource build(BusTopicParams busTopicParams) {
-        if (busTopicParams.getServers() == null || busTopicParams.getServers().isEmpty()) {
-            throw new IllegalArgumentException("UEB Server(s) must be provided");
-        }
-
-        if (busTopicParams.getTopic() == null || busTopicParams.getTopic().isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (uebTopicSources.containsKey(busTopicParams.getTopic())) {
-                return uebTopicSources.get(busTopicParams.getTopic());
-            }
-
-            UebTopicSource uebTopicSource = makeSource(busTopicParams);
-
-            if (busTopicParams.isManaged()) {
-                uebTopicSources.put(busTopicParams.getTopic(), uebTopicSource);
-            }
-
-            return uebTopicSource;
-        }
-    }
-
-    @Override
-    public List<UebTopicSource> build(Properties properties) {
-
-        String readTopics = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS);
-        if (readTopics == null || readTopics.isEmpty()) {
-            logger.info("{}: no topic for UEB Source", this);
-            return new ArrayList<>();
-        }
-        List<String> readTopicList = new ArrayList<>(Arrays.asList(readTopics.split("\\s*,\\s*")));
-
-        List<UebTopicSource> newUebTopicSources = new ArrayList<>();
-        synchronized (this) {
-            for (String topic : readTopicList) {
-                if (this.uebTopicSources.containsKey(topic)) {
-                    newUebTopicSources.add(this.uebTopicSources.get(topic));
-                    continue;
-                }
-
-                String servers = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SERVERS_SUFFIX);
-
-                if (servers == null || servers.isEmpty()) {
-                    logger.error("{}: no UEB servers configured for sink {}", this, topic);
-                    continue;
-                }
-
-                final List<String> serverList = new ArrayList<>(Arrays.asList(servers.split("\\s*,\\s*")));
-
-                final String apiKey = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_KEY_SUFFIX);
-
-                final String apiSecret = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_API_SECRET_SUFFIX);
-
-                final String consumerGroup = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS 
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_GROUP_SUFFIX);
-
-                final String consumerInstance = properties.getProperty(
-                        PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_CONSUMER_INSTANCE_SUFFIX);
-
-                String fetchTimeoutString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_TIMEOUT_SUFFIX);
-                int fetchTimeout = UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH;
-                if (fetchTimeoutString != null && !fetchTimeoutString.isEmpty()) {
-                    try {
-                        fetchTimeout = Integer.parseInt(fetchTimeoutString);
-                    } catch (NumberFormatException nfe) {
-                        logger.warn("{}: fetch timeout {} is in invalid format for topic {} ", this, fetchTimeoutString,
-                                topic);
-                    }
-                }
-
-                String fetchLimitString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS
-                        + "." + topic + PolicyEndPointProperties.PROPERTY_TOPIC_SOURCE_FETCH_LIMIT_SUFFIX);
-                int fetchLimit = UebTopicSource.DEFAULT_LIMIT_FETCH;
-                if (fetchLimitString != null && !fetchLimitString.isEmpty()) {
-                    try {
-                        fetchLimit = Integer.parseInt(fetchLimitString);
-                    } catch (NumberFormatException nfe) {
-                        logger.warn("{}: fetch limit {} is in invalid format for topic {} ", this, fetchLimitString,
-                                topic);
-                    }
-                }
-
-                String managedString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_MANAGED_SUFFIX);
-                boolean managed = true;
-                if (managedString != null && !managedString.isEmpty()) {
-                    managed = Boolean.parseBoolean(managedString);
-                }
-
-                String useHttpsString = properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "."
-                        + topic + PolicyEndPointProperties.PROPERTY_HTTP_HTTPS_SUFFIX);
-
-                // default is to use HTTP if no https property exists
-                boolean useHttps = false;
-                if (useHttpsString != null && !useHttpsString.isEmpty()) {
-                    useHttps = Boolean.parseBoolean(useHttpsString);
-                }
-
-                String allowSelfSignedCertsString =
-                        properties.getProperty(PolicyEndPointProperties.PROPERTY_UEB_SOURCE_TOPICS + "." + topic
-                                + PolicyEndPointProperties.PROPERTY_ALLOW_SELF_SIGNED_CERTIFICATES_SUFFIX);
-
-                // default is to disallow self-signed certs
-                boolean allowSelfSignedCerts = false;
-                if (allowSelfSignedCertsString != null && !allowSelfSignedCertsString.isEmpty()) {
-                    allowSelfSignedCerts = Boolean.parseBoolean(allowSelfSignedCertsString);
-                }
-
-                UebTopicSource uebTopicSource = this.build(BusTopicParams.builder()
-                        .servers(serverList)
-                        .topic(topic)
-                        .apiKey(apiKey)
-                        .apiSecret(apiSecret)
-                        .consumerGroup(consumerGroup)
-                        .consumerInstance(consumerInstance)
-                        .fetchTimeout(fetchTimeout)
-                        .fetchLimit(fetchLimit)
-                        .managed(managed)
-                        .useHttps(useHttps)
-                        .allowSelfSignedCerts(allowSelfSignedCerts).build());
-                newUebTopicSources.add(uebTopicSource);
-            }
-        }
-        return newUebTopicSources;
-    }
-
-    @Override
-    public UebTopicSource build(List<String> servers, String topic, String apiKey, String apiSecret) {
-
-        return this.build(BusTopicParams.builder()
-                .servers(servers)
-                .topic(topic)
-                .apiKey(apiKey)
-                .apiSecret(apiSecret)
-                .fetchTimeout(UebTopicSource.DEFAULT_TIMEOUT_MS_FETCH)
-                .fetchLimit(UebTopicSource.DEFAULT_LIMIT_FETCH)
-                .managed(true)
-                .useHttps(false)
-                .allowSelfSignedCerts(true).build());
-    }
-
-    @Override
-    public UebTopicSource build(List<String> servers, String topic) {
-        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) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        UebTopicSource uebTopicSource;
-
-        synchronized (this) {
-            if (!uebTopicSources.containsKey(topic)) {
-                return;
-            }
-
-            uebTopicSource = uebTopicSources.remove(topic);
-        }
-
-        uebTopicSource.shutdown();
-    }
-
-    @Override
-    public void destroy() {
-        List<UebTopicSource> readers = this.inventory();
-        for (UebTopicSource reader : readers) {
-            reader.shutdown();
-        }
-
-        synchronized (this) {
-            this.uebTopicSources.clear();
-        }
-    }
-
-    @Override
-    public UebTopicSource get(String topic) {
-
-        if (topic == null || topic.isEmpty()) {
-            throw new IllegalArgumentException(MISSING_TOPIC);
-        }
-
-        synchronized (this) {
-            if (uebTopicSources.containsKey(topic)) {
-                return uebTopicSources.get(topic);
-            } else {
-                throw new IllegalStateException("UebTopiceSource for " + topic + " not found");
-            }
-        }
-    }
-
-    @Override
-    public synchronized List<UebTopicSource> inventory() {
-        return new ArrayList<>(this.uebTopicSources.values());
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder builder = new StringBuilder();
-        builder.append("IndexedUebTopicSourceFactory []");
-        return builder.toString();
-    }
-}