22 files changed, 2468 insertions, 2468 deletions
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/DMaaPMRFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
index a23c1e9..7163fe2 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/DMaaPMRFactory.java
@@ -1,112 +1,112 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap;
-
-import com.google.inject.AbstractModule;
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.module.AnalyticsDMaaPModule;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nonnull;
-
-/**
- * Creates pre injected implementations for {@link DMaaPMRPublisher} and {@link DMaaPMRSubscriber}
- * <p>
- *     Usage:
- *     <p>Create an instance of DMaaP MR Factory</p>
- *     <pre>
- *        DMaaPFactory dmaapFactory = DMaaPFactory.initalize()
- *     </pre>
- *     <p>Create a new DMaaP MR Publisher</p>
- *     <pre>
- *         DMaaPMRPublisher publisher = dmaapFactory.createPublisher(publisherConfig)
- *     </pre>
- *     <p>Create new DMaaP MR Subscriber</p>
- *     <pre>
- *         DMaaPMRSubscriber subscriber = dmaapFactory.createSubscriber(subscriberConfig)
- *     </pre>
- * <p>
- * <strong>All Clients must use this Factory to initalize DMaaP Message Router Publishers and Subscribers</strong>
- * </p>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public class DMaaPMRFactory {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRFactory.class);
-
-    private final Injector injector;
-
-    public DMaaPMRFactory(AbstractModule guiceModule) {
-        injector = Guice.createInjector(guiceModule);
-    }
-
-    /**
-     * Returns configured instance of {@link DMaaPMRPublisher}
-     *
-     * @param publisherConfig Publisher Config
-     * @return configured instance of DMaaP MR Publisher
-     */
-    public DMaaPMRPublisher createPublisher(@Nonnull DMaaPMRPublisherConfig publisherConfig) {
-        final DMaaPMRPublisherFactory publisherFactory = injector.getInstance(DMaaPMRPublisherFactory.class);
-        LOG.debug("Creating new DMaaP MR Publisher Instance with configuration: {}", publisherConfig);
-        final DMaaPMRPublisher dMaaPMRPublisher = publisherFactory.create(publisherConfig);
-        LOG.info("Created new DMaaP MR Publisher Instance. Publisher creation time: {}",
-                dMaaPMRPublisher.getPublisherCreationTime());
-        return dMaaPMRPublisher;
-    }
-
-    /**
-     * Returns configured instance of {@link DMaaPMRSubscriber}
-     *
-     * @param subscriberConfig Subscriber Config
-     * @return configured instance of DMaaP MR Subscriber
-     */
-    public DMaaPMRSubscriber createSubscriber(@Nonnull DMaaPMRSubscriberConfig subscriberConfig) {
-        final DMaaPMRSubscriberFactory subscriberFactory = injector.getInstance(DMaaPMRSubscriberFactory.class);
-        LOG.debug("Creating new DMaaP MR Subscriber Instance with configuration: {}", subscriberConfig);
-        final DMaaPMRSubscriber dMaaPMRSubscriber = subscriberFactory.create(subscriberConfig);
-        LOG.info("Created new DMaaP MR Subscriber Instance. Subscriber creation time: {}",
-                dMaaPMRSubscriber.getSubscriberCreationTime());
-        return dMaaPMRSubscriber;
-    }
-
-    /**
-     * Creates an instance of {@link DMaaPMRFactory}
-     *
-     * @return {@link DMaaPMRFactory} factory instance
-     */
-    public static DMaaPMRFactory create() {
-        final DMaaPMRFactory dMaaPMRFactory = new DMaaPMRFactory(new AnalyticsDMaaPModule());
-        LOG.info("Created new instance of DMaaP MR Factory");
-        return dMaaPMRFactory;
-    }
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.module.AnalyticsDMaaPModule;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+/**
+ * Creates pre injected implementations for {@link DMaaPMRPublisher} and {@link DMaaPMRSubscriber}
+ * <p>
+ *     Usage:
+ *     <p>Create an instance of DMaaP MR Factory</p>
+ *     <pre>
+ *        DMaaPFactory dmaapFactory = DMaaPFactory.initalize()
+ *     </pre>
+ *     <p>Create a new DMaaP MR Publisher</p>
+ *     <pre>
+ *         DMaaPMRPublisher publisher = dmaapFactory.createPublisher(publisherConfig)
+ *     </pre>
+ *     <p>Create new DMaaP MR Subscriber</p>
+ *     <pre>
+ *         DMaaPMRSubscriber subscriber = dmaapFactory.createSubscriber(subscriberConfig)
+ *     </pre>
+ * <p>
+ * <strong>All Clients must use this Factory to initalize DMaaP Message Router Publishers and Subscribers</strong>
+ * </p>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public class DMaaPMRFactory {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRFactory.class);
+
+    private final Injector injector;
+
+    public DMaaPMRFactory(AbstractModule guiceModule) {
+        injector = Guice.createInjector(guiceModule);
+    }
+
+    /**
+     * Returns configured instance of {@link DMaaPMRPublisher}
+     *
+     * @param publisherConfig Publisher Config
+     * @return configured instance of DMaaP MR Publisher
+     */
+    public DMaaPMRPublisher createPublisher(@Nonnull DMaaPMRPublisherConfig publisherConfig) {
+        final DMaaPMRPublisherFactory publisherFactory = injector.getInstance(DMaaPMRPublisherFactory.class);
+        LOG.debug("Creating new DMaaP MR Publisher Instance with configuration: {}", publisherConfig);
+        final DMaaPMRPublisher dMaaPMRPublisher = publisherFactory.create(publisherConfig);
+        LOG.info("Created new DMaaP MR Publisher Instance. Publisher creation time: {}",
+                dMaaPMRPublisher.getPublisherCreationTime());
+        return dMaaPMRPublisher;
+    }
+
+    /**
+     * Returns configured instance of {@link DMaaPMRSubscriber}
+     *
+     * @param subscriberConfig Subscriber Config
+     * @return configured instance of DMaaP MR Subscriber
+     */
+    public DMaaPMRSubscriber createSubscriber(@Nonnull DMaaPMRSubscriberConfig subscriberConfig) {
+        final DMaaPMRSubscriberFactory subscriberFactory = injector.getInstance(DMaaPMRSubscriberFactory.class);
+        LOG.debug("Creating new DMaaP MR Subscriber Instance with configuration: {}", subscriberConfig);
+        final DMaaPMRSubscriber dMaaPMRSubscriber = subscriberFactory.create(subscriberConfig);
+        LOG.info("Created new DMaaP MR Subscriber Instance. Subscriber creation time: {}",
+                dMaaPMRSubscriber.getSubscriberCreationTime());
+        return dMaaPMRSubscriber;
+    }
+
+    /**
+     * Creates an instance of {@link DMaaPMRFactory}
+     *
+     * @return {@link DMaaPMRFactory} factory instance
+     */
+    public static DMaaPMRFactory create() {
+        final DMaaPMRFactory dMaaPMRFactory = new DMaaPMRFactory(new AnalyticsDMaaPModule());
+        LOG.info("Created new instance of DMaaP MR Factory");
+        return dMaaPMRFactory;
+    }
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
index 32a4357..3f74091 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRBaseConfig.java
@@ -1,193 +1,193 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Locale;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.JSON_APPLICATION_TYPE;
-
-/**
- * <p>
- *      Contains common parameters for both DMaaP Message Router Publisher and Subscriber Configs
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public abstract class DMaaPMRBaseConfig implements DMaaPMRConfig {
-
-    protected static final Logger LOG = LoggerFactory.getLogger(DMaaPMRBaseConfig.class);
-
-    protected String hostName;
-    protected Integer portNumber;
-    protected String topicName;
-    protected String protocol;
-    protected String userName;
-    protected String userPassword;
-    protected String contentType;
-
-    /**
-     * Provides host name e.g. mrlocal-mtnjftle01.homer.com
-     *
-     * @return host name
-     */
-    public String getHostName() {
-        return hostName;
-    }
-
-
-    /**
-     * Provides Port Number of DMaaP MR Topic Host. Defaults to 80
-     *
-     * @return host port number
-     */
-    public Integer getPortNumber() {
-        return portNumber;
-    }
-
-    /**
-     * Provides topic name e.g. com.dcae.dmaap.mtnje2.DcaeTestVES
-     *
-     * @return topic name
-     */
-    public String getTopicName() {
-        return topicName;
-    }
-
-    /**
-     * Provides protocol type e.g. http or https
-     *
-     * @return protocol type
-     */
-    public String getProtocol() {
-        return protocol;
-    }
-
-    /**
-     * Provides content type e.g. application/json
-     *
-     * @return content type
-     */
-    public String getContentType() {
-        return contentType;
-    }
-
-
-    /**
-     * Provides User name for the DMaaP MR Topic authentication
-     *
-     * @return user name
-     */
-    public String getUserName() {
-        return userName;
-    }
-
-    /**
-     * Provides User password for the DMaaP MR Topic authentication
-     *
-     * @return user Password
-     */
-    public String getUserPassword() {
-        return userPassword;
-    }
-
-
-    /**
-     * Trims, adjusts casing and validates user input String for protocol selection
-     *
-     * @param protocol - User input for protocol String
-     * @return - network protocol e.g http or https
-     */
-    protected static String normalizeValidateProtocol(final String protocol) {
-        // validate that only http and https are supported protocols are Supported for DMaaP MR
-        String normalizedProtocolString = protocol.trim().toLowerCase(Locale.ENGLISH);
-        if (normalizedProtocolString.isEmpty() ||
-                !("http".equals(normalizedProtocolString) || "https".equals(normalizedProtocolString))) {
-
-            final String errorMessage =
-                    "Unsupported protocol selection. Only HTTPS and HTTPS are currently supported for DMaaP MR";
-
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
-        }
-        return normalizedProtocolString;
-    }
-
-
-    /**
-     * Trims, adjust casing and validates content type is supported by DMaaP.
-     *
-     * NOTE: DMaaP currently only support application/json content type
-     *
-     * @param contentType content type that needs to checked for DMaaP MR support
-     * @return true if content type is supported by DMaaP MR
-     */
-    protected static String normalizeValidateContentType(final String contentType) {
-        // Current DMaaP MR is only supporting "application/json" content type
-        String normalizedContentType = contentType.trim().toLowerCase(Locale.ENGLISH);
-        final boolean isSupported = contentType.equals(JSON_APPLICATION_TYPE);
-        if (!isSupported) {
-            final String errorMessage =
-                    "Unsupported content type selection. Only application/json is currently supported for DMaaP MR";
-
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
-        }
-        return normalizedContentType;
-    }
-
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (!(o instanceof DMaaPMRBaseConfig)) {
-            return false;
-        }
-        DMaaPMRBaseConfig that = (DMaaPMRBaseConfig) o;
-        return Objects.equal(hostName, that.hostName) &&
-                Objects.equal(portNumber, that.portNumber) &&
-                Objects.equal(topicName, that.topicName) &&
-                Objects.equal(protocol, that.protocol) &&
-                Objects.equal(userName, that.userName) &&
-                Objects.equal(userPassword, that.userPassword) &&
-                Objects.equal(contentType, that.contentType);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hashCode(hostName, portNumber, topicName, protocol, userName, userPassword, contentType);
-    }
-
-    @Override
-    public String toString() {
-        return Objects.toStringHelper(this)
-                .add("hostName", hostName)
-                .add("portNumber", portNumber)
-                .add("topicName", topicName)
-                .add("protocol", protocol)
-                .add("userName", userName)
-                .add("contentType", contentType)
-                .toString();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Locale;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.JSON_APPLICATION_TYPE;
+
+/**
+ * <p>
+ *      Contains common parameters for both DMaaP Message Router Publisher and Subscriber Configs
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public abstract class DMaaPMRBaseConfig implements DMaaPMRConfig {
+
+    protected static final Logger LOG = LoggerFactory.getLogger(DMaaPMRBaseConfig.class);
+
+    protected String hostName;
+    protected Integer portNumber;
+    protected String topicName;
+    protected String protocol;
+    protected String userName;
+    protected String userPassword;
+    protected String contentType;
+
+    /**
+     * Provides host name e.g. mrlocal-mtnjftle01.homer.com
+     *
+     * @return host name
+     */
+    public String getHostName() {
+        return hostName;
+    }
+
+
+    /**
+     * Provides Port Number of DMaaP MR Topic Host. Defaults to 80
+     *
+     * @return host port number
+     */
+    public Integer getPortNumber() {
+        return portNumber;
+    }
+
+    /**
+     * Provides topic name e.g. com.dcae.dmaap.mtnje2.DcaeTestVES
+     *
+     * @return topic name
+     */
+    public String getTopicName() {
+        return topicName;
+    }
+
+    /**
+     * Provides protocol type e.g. http or https
+     *
+     * @return protocol type
+     */
+    public String getProtocol() {
+        return protocol;
+    }
+
+    /**
+     * Provides content type e.g. application/json
+     *
+     * @return content type
+     */
+    public String getContentType() {
+        return contentType;
+    }
+
+
+    /**
+     * Provides User name for the DMaaP MR Topic authentication
+     *
+     * @return user name
+     */
+    public String getUserName() {
+        return userName;
+    }
+
+    /**
+     * Provides User password for the DMaaP MR Topic authentication
+     *
+     * @return user Password
+     */
+    public String getUserPassword() {
+        return userPassword;
+    }
+
+
+    /**
+     * Trims, adjusts casing and validates user input String for protocol selection
+     *
+     * @param protocol - User input for protocol String
+     * @return - network protocol e.g http or https
+     */
+    protected static String normalizeValidateProtocol(final String protocol) {
+        // validate that only http and https are supported protocols are Supported for DMaaP MR
+        String normalizedProtocolString = protocol.trim().toLowerCase(Locale.ENGLISH);
+        if (normalizedProtocolString.isEmpty() ||
+                !("http".equals(normalizedProtocolString) || "https".equals(normalizedProtocolString))) {
+
+            final String errorMessage =
+                    "Unsupported protocol selection. Only HTTPS and HTTPS are currently supported for DMaaP MR";
+
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
+        }
+        return normalizedProtocolString;
+    }
+
+
+    /**
+     * Trims, adjust casing and validates content type is supported by DMaaP.
+     *
+     * NOTE: DMaaP currently only support application/json content type
+     *
+     * @param contentType content type that needs to checked for DMaaP MR support
+     * @return true if content type is supported by DMaaP MR
+     */
+    protected static String normalizeValidateContentType(final String contentType) {
+        // Current DMaaP MR is only supporting "application/json" content type
+        String normalizedContentType = contentType.trim().toLowerCase(Locale.ENGLISH);
+        final boolean isSupported = contentType.equals(JSON_APPLICATION_TYPE);
+        if (!isSupported) {
+            final String errorMessage =
+                    "Unsupported content type selection. Only application/json is currently supported for DMaaP MR";
+
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
+        }
+        return normalizedContentType;
+    }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof DMaaPMRBaseConfig)) {
+            return false;
+        }
+        DMaaPMRBaseConfig that = (DMaaPMRBaseConfig) o;
+        return Objects.equal(hostName, that.hostName) &&
+                Objects.equal(portNumber, that.portNumber) &&
+                Objects.equal(topicName, that.topicName) &&
+                Objects.equal(protocol, that.protocol) &&
+                Objects.equal(userName, that.userName) &&
+                Objects.equal(userPassword, that.userPassword) &&
+                Objects.equal(contentType, that.contentType);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(hostName, portNumber, topicName, protocol, userName, userPassword, contentType);
+    }
+
+    @Override
+    public String toString() {
+        return Objects.toStringHelper(this)
+                .add("hostName", hostName)
+                .add("portNumber", portNumber)
+                .add("topicName", topicName)
+                .add("protocol", protocol)
+                .add("userName", userName)
+                .add("contentType", contentType)
+                .toString();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
index 687d489..2ff3c4e 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRConfig.java
@@ -1,32 +1,32 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-/**
- * <p>
- *      Marker Interface for all DMaaP MR Configs.
- *      Holds Default configuration parameters for all Configs
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public interface DMaaPMRConfig {
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.config;
+
+/**
+ * <p>
+ *      Marker Interface for all DMaaP MR Configs.
+ *      Holds Default configuration parameters for all Configs
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public interface DMaaPMRConfig {
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
index 6fbebf2..8dfc289 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRPublisherConfig.java
@@ -1,248 +1,248 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- *      Immutable DMaaP MR Configuration for DMaaP MR Publisher.
- * <p>
- *      Use {@link DMaaPMRPublisherConfig.Builder} to construct Subscriber Configuration
- * </p>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public class DMaaPMRPublisherConfig extends DMaaPMRBaseConfig {
-
-    /**
-     * Publisher batching queue size
-     */
-    private int maxBatchSize;
-
-    /**
-     * Publisher Recovery Queue Size
-     */
-    private int maxRecoveryQueueSize;
-
-
-    private DMaaPMRPublisherConfig(@Nonnull String hostName,
-                                   @Nonnull Integer portNumber,
-                                   @Nonnull String topicName,
-                                   @Nonnull String protocol,
-                                   String userName,
-                                   String userPassword,
-                                   @Nonnull String contentType,
-                                   int maxBatchSize,
-                                   int maxRecoveryQueueSize) {
-        this.hostName = hostName;
-        this.portNumber = portNumber;
-        this.topicName = topicName;
-        this.protocol = protocol;
-        this.userName = userName;
-        this.userPassword = userPassword;
-        this.contentType = contentType;
-        this.maxBatchSize = maxBatchSize;
-        this.maxRecoveryQueueSize = maxRecoveryQueueSize;
-    }
-
-
-    /**
-     * Builder to initialize immutable {@link DMaaPMRPublisherConfig} object
-     */
-    public static class Builder {
-
-        private String hostName;
-        private Integer portNumber;
-        private String topicName;
-        private String userName;
-        private String userPassword;
-        private String protocol;
-        private String contentType;
-        private int maxBatchSize;
-        private int maxRecoveryQueueSize;
-
-        public Builder(@Nonnull String hostName, @Nonnull String topicName) {
-            // required values
-            this.hostName = hostName;
-            this.topicName = topicName;
-            // Default values
-            this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
-            this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
-            this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
-            this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
-            this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
-            this.maxBatchSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_BATCH_SIZE;
-            this.maxRecoveryQueueSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
-        }
-
-        /**
-         * Setup for custom host port number - Defaults to 80.
-         *
-         * @param portNumber custom port number
-         * @return Builder object itself for chaining
-         */
-        public Builder setPortNumber(@Nonnull Integer portNumber) {
-            this.portNumber = portNumber;
-            return this;
-        }
-
-
-        /**
-         * Setup user name for authentication. If no username is provided authentication will be disabled
-         *
-         * @param userName user name for DMaaP Topic Authentication
-         * @return Builder object itself for chaining
-         */
-        public Builder setUserName(@Nonnull String userName) {
-            this.userName = userName;
-            return this;
-        }
-
-
-        /**
-         * Setup user password for authentication. If no password is provided authentication will be disabled
-         *
-         * @param userPassword user password for DMaaP Topic Authentication
-         * @return Builder object itself for chaining
-         */
-        public Builder setUserPassword(@Nonnull String userPassword) {
-            this.userPassword = userPassword;
-            return this;
-        }
-
-
-        /**
-         * Setup custom Publisher protocol - Defaults to https.
-         * Note: Only http and https are currently supported.
-         *
-         * @param protocol protocol e.g. https
-         * @return Builder object itself for chaining
-         */
-        public Builder setProtocol(@Nonnull String protocol) {
-            this.protocol = normalizeValidateProtocol(protocol);
-            return this;
-        }
-
-
-        /**
-         * Setup custom Publisher content-type - Defaults to application/json
-         *
-         * @param contentType content type e.g. application/json
-         * @return Builder object itself for chaining
-         */
-        public Builder setContentType(@Nonnull String contentType) {
-            final String normalizedContentType = normalizeValidateContentType(contentType);
-            this.contentType = normalizedContentType;
-            return this;
-        }
-
-
-        /**
-         * Setup custom Publisher Max Batch Size - Defaults to 100
-         *
-         * @param maxBatchSize max Batch Size
-         * @return Builder object itself for chaining
-         */
-        public Builder setMaxBatchSize(int maxBatchSize) {
-            this.maxBatchSize = maxBatchSize;
-            return this;
-        }
-
-
-        /**
-         * Setup custom Maximum Recovery Queue Size. Recovery Queue is used to hold messages temporarily in case
-         * DMaaP MR Publisher topic is not responding for any reason. Defaults to 100,000
-         *
-         * @param maxRecoveryQueueSize max recovery queue size
-         * @return Builder object itself for chaining
-         */
-        public Builder setMaxRecoveryQueueSize(int maxRecoveryQueueSize) {
-            this.maxRecoveryQueueSize = maxRecoveryQueueSize;
-            return this;
-        }
-
-        /**
-         * Creates immutable instance of {@link DMaaPMRPublisherConfig}
-         *
-         * @return Builds and returns thread safe, immutable {@link DMaaPMRPublisherConfig} object
-         */
-        public DMaaPMRPublisherConfig build() {
-            return new DMaaPMRPublisherConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
-                    contentType, maxBatchSize, maxRecoveryQueueSize);
-        }
-
-    }
-
-
-    /**
-     * Returns max Publisher Batch Queue Size
-     *
-     * @return max Publisher Batch Queue size
-     */
-    public int getMaxBatchSize() {
-        return maxBatchSize;
-    }
-
-    /**
-     * Returns max Publisher Recovery Queue Size
-     *
-     * @return max Recovery Queue size
-     */
-    public int getMaxRecoveryQueueSize() {
-        return maxRecoveryQueueSize;
-    }
-
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        if (!super.equals(o)) {
-            return false;
-        }
-        DMaaPMRPublisherConfig that = (DMaaPMRPublisherConfig) o;
-        return maxBatchSize == that.maxBatchSize &&
-                maxRecoveryQueueSize == that.maxRecoveryQueueSize;
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hashCode(super.hashCode(), maxBatchSize, maxRecoveryQueueSize);
-    }
-
-
-    @Override
-    public String toString() {
-        return Objects.toStringHelper(this)
-                .add("baseConfig", super.toString())
-                .add("maxBatchSize", maxBatchSize)
-                .add("maxRecoveryQueueSize", maxRecoveryQueueSize)
-                .toString();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ *      Immutable DMaaP MR Configuration for DMaaP MR Publisher.
+ * <p>
+ *      Use {@link DMaaPMRPublisherConfig.Builder} to construct Subscriber Configuration
+ * </p>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public class DMaaPMRPublisherConfig extends DMaaPMRBaseConfig {
+
+    /**
+     * Publisher batching queue size
+     */
+    private int maxBatchSize;
+
+    /**
+     * Publisher Recovery Queue Size
+     */
+    private int maxRecoveryQueueSize;
+
+
+    private DMaaPMRPublisherConfig(@Nonnull String hostName,
+                                   @Nonnull Integer portNumber,
+                                   @Nonnull String topicName,
+                                   @Nonnull String protocol,
+                                   String userName,
+                                   String userPassword,
+                                   @Nonnull String contentType,
+                                   int maxBatchSize,
+                                   int maxRecoveryQueueSize) {
+        this.hostName = hostName;
+        this.portNumber = portNumber;
+        this.topicName = topicName;
+        this.protocol = protocol;
+        this.userName = userName;
+        this.userPassword = userPassword;
+        this.contentType = contentType;
+        this.maxBatchSize = maxBatchSize;
+        this.maxRecoveryQueueSize = maxRecoveryQueueSize;
+    }
+
+
+    /**
+     * Builder to initialize immutable {@link DMaaPMRPublisherConfig} object
+     */
+    public static class Builder {
+
+        private String hostName;
+        private Integer portNumber;
+        private String topicName;
+        private String userName;
+        private String userPassword;
+        private String protocol;
+        private String contentType;
+        private int maxBatchSize;
+        private int maxRecoveryQueueSize;
+
+        public Builder(@Nonnull String hostName, @Nonnull String topicName) {
+            // required values
+            this.hostName = hostName;
+            this.topicName = topicName;
+            // Default values
+            this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
+            this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
+            this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
+            this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
+            this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
+            this.maxBatchSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_BATCH_SIZE;
+            this.maxRecoveryQueueSize = AnalyticsConstants.DEFAULT_PUBLISHER_MAX_RECOVERY_QUEUE_SIZE;
+        }
+
+        /**
+         * Setup for custom host port number - Defaults to 80.
+         *
+         * @param portNumber custom port number
+         * @return Builder object itself for chaining
+         */
+        public Builder setPortNumber(@Nonnull Integer portNumber) {
+            this.portNumber = portNumber;
+            return this;
+        }
+
+
+        /**
+         * Setup user name for authentication. If no username is provided authentication will be disabled
+         *
+         * @param userName user name for DMaaP Topic Authentication
+         * @return Builder object itself for chaining
+         */
+        public Builder setUserName(@Nonnull String userName) {
+            this.userName = userName;
+            return this;
+        }
+
+
+        /**
+         * Setup user password for authentication. If no password is provided authentication will be disabled
+         *
+         * @param userPassword user password for DMaaP Topic Authentication
+         * @return Builder object itself for chaining
+         */
+        public Builder setUserPassword(@Nonnull String userPassword) {
+            this.userPassword = userPassword;
+            return this;
+        }
+
+
+        /**
+         * Setup custom Publisher protocol - Defaults to https.
+         * Note: Only http and https are currently supported.
+         *
+         * @param protocol protocol e.g. https
+         * @return Builder object itself for chaining
+         */
+        public Builder setProtocol(@Nonnull String protocol) {
+            this.protocol = normalizeValidateProtocol(protocol);
+            return this;
+        }
+
+
+        /**
+         * Setup custom Publisher content-type - Defaults to application/json
+         *
+         * @param contentType content type e.g. application/json
+         * @return Builder object itself for chaining
+         */
+        public Builder setContentType(@Nonnull String contentType) {
+            final String normalizedContentType = normalizeValidateContentType(contentType);
+            this.contentType = normalizedContentType;
+            return this;
+        }
+
+
+        /**
+         * Setup custom Publisher Max Batch Size - Defaults to 100
+         *
+         * @param maxBatchSize max Batch Size
+         * @return Builder object itself for chaining
+         */
+        public Builder setMaxBatchSize(int maxBatchSize) {
+            this.maxBatchSize = maxBatchSize;
+            return this;
+        }
+
+
+        /**
+         * Setup custom Maximum Recovery Queue Size. Recovery Queue is used to hold messages temporarily in case
+         * DMaaP MR Publisher topic is not responding for any reason. Defaults to 100,000
+         *
+         * @param maxRecoveryQueueSize max recovery queue size
+         * @return Builder object itself for chaining
+         */
+        public Builder setMaxRecoveryQueueSize(int maxRecoveryQueueSize) {
+            this.maxRecoveryQueueSize = maxRecoveryQueueSize;
+            return this;
+        }
+
+        /**
+         * Creates immutable instance of {@link DMaaPMRPublisherConfig}
+         *
+         * @return Builds and returns thread safe, immutable {@link DMaaPMRPublisherConfig} object
+         */
+        public DMaaPMRPublisherConfig build() {
+            return new DMaaPMRPublisherConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
+                    contentType, maxBatchSize, maxRecoveryQueueSize);
+        }
+
+    }
+
+
+    /**
+     * Returns max Publisher Batch Queue Size
+     *
+     * @return max Publisher Batch Queue size
+     */
+    public int getMaxBatchSize() {
+        return maxBatchSize;
+    }
+
+    /**
+     * Returns max Publisher Recovery Queue Size
+     *
+     * @return max Recovery Queue size
+     */
+    public int getMaxRecoveryQueueSize() {
+        return maxRecoveryQueueSize;
+    }
+
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        DMaaPMRPublisherConfig that = (DMaaPMRPublisherConfig) o;
+        return maxBatchSize == that.maxBatchSize &&
+                maxRecoveryQueueSize == that.maxRecoveryQueueSize;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(super.hashCode(), maxBatchSize, maxRecoveryQueueSize);
+    }
+
+
+    @Override
+    public String toString() {
+        return Objects.toStringHelper(this)
+                .add("baseConfig", super.toString())
+                .add("maxBatchSize", maxBatchSize)
+                .add("maxRecoveryQueueSize", maxRecoveryQueueSize)
+                .toString();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
index 7252ce0..8158b6b 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/config/DMaaPMRSubscriberConfig.java
@@ -1,295 +1,295 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.config;
-
-import com.google.common.base.Objects;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-
-import java.util.UUID;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- *      Immutable DMaaP MR Configuration for Subscriber.
- * <p>
- *      Use {@link DMaaPMRSubscriberConfig.Builder} to construct Subscriber Configuration
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/12/2016.
- */
-public final class DMaaPMRSubscriberConfig extends DMaaPMRBaseConfig {
-
-    private final String consumerId;
-    private final String consumerGroup;
-    private final Integer timeoutMS;
-    private final Integer messageLimit;
-
-    private DMaaPMRSubscriberConfig(@Nonnull String hostName,
-                                    @Nonnull Integer portNumber,
-                                    @Nonnull String topicName,
-                                    @Nonnull String protocol,
-                                    String userName,
-                                    String userPassword,
-                                    @Nonnull String contentType,
-                                    @Nonnull String consumerId,
-                                    @Nonnull String consumerGroup,
-                                    @Nonnull Integer timeoutMS,
-                                    @Nonnull Integer messageLimit) {
-        this.hostName = hostName;
-        this.portNumber = portNumber;
-        this.topicName = topicName;
-        this.protocol = protocol;
-        this.userName = userName;
-        this.userPassword = userPassword;
-        this.contentType = contentType;
-        this.consumerId = consumerId;
-        this.consumerGroup = consumerGroup;
-        this.timeoutMS = timeoutMS;
-        this.messageLimit = messageLimit;
-    }
-
-    /**
-     * Builder to initialize immutable {@link DMaaPMRSubscriberConfig} object
-     */
-    public static class Builder {
-
-        private String hostName;
-        private Integer portNumber;
-        private String topicName;
-        private String userName;
-        private String userPassword;
-        private String protocol;
-        private String contentType;
-        private String consumerId;
-        private String consumerGroup;
-        private Integer timeoutMS;
-        private Integer messageLimit;
-
-        public Builder(@Nonnull String hostName,
-                       @Nonnull String topicName) {
-            // Required Values
-            this.hostName = hostName;
-            this.topicName = topicName;
-
-            // Default values
-            this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
-            this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
-            this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
-            this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
-            this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
-            this.consumerId = UUID.randomUUID().toString(); // consumer is assigned a random id by default
-            this.consumerGroup = AnalyticsConstants.DEFAULT_SUBSCRIBER_GROUP_PREFIX + consumerId; // random group
-            this.timeoutMS = AnalyticsConstants.DEFAULT_SUBSCRIBER_TIMEOUT_MS; // defaults to 10ms timeout
-            this.messageLimit = AnalyticsConstants.DEFAULT_SUBSCRIBER_MESSAGE_LIMIT; // defaults to 1000 message limit
-        }
-
-
-        /**
-         * Setup for custom host port number - Defaults to 80.
-         *
-         * @param portNumber custom port number
-         * @return Builder object itself for chaining
-         */
-        public Builder setPortNumber(@Nonnull Integer portNumber) {
-            this.portNumber = portNumber;
-            return this;
-        }
-
-
-        /**
-         * Setup user name for authentication. If no username is provided authentication will be disabled
-         *
-         * @param userName user name for DMaaP Topic Authentication
-         * @return Builder object itself for chaining
-         */
-        public Builder setUserName(@Nonnull String userName) {
-            this.userName = userName;
-            return this;
-        }
-
-
-        /**
-         * Setup user password for authentication. If no password is provided authentication will be disabled
-         *
-         * @param userPassword user password for DMaaP Topic Authentication
-         * @return Builder object itself for chaining
-         */
-        public Builder setUserPassword(@Nonnull String userPassword) {
-            this.userPassword = userPassword;
-            return this;
-        }
-
-
-        /**
-         * Setup custom Subscriber protocol - Defaults to https.
-         * Note: Only http and https are currently supported.
-         *
-         * @param protocol protocol e.g. https or http
-         * @return Builder object itself for chaining
-         */
-        public Builder setProtocol(@Nonnull String protocol) {
-
-            this.protocol = normalizeValidateProtocol(protocol);
-            return this;
-        }
-
-        /**
-         * Setup custom Subscriber content-type - Defaults to application/json
-         *
-         * @param contentType content type e.g. application/json
-         * @return Builder object itself for chaining
-         */
-        public Builder setContentType(@Nonnull String contentType) {
-            final String normalizedContentType = normalizeValidateContentType(contentType);
-            this.contentType = normalizedContentType;
-            return this;
-        }
-
-
-        /**
-         * Setup custom Consumer Id - Defaults to random Id
-         *
-         * @param consumerId - custom consumer ID
-         * @return Builder object itself for chaining
-         */
-        public Builder setConsumerId(@Nonnull String consumerId) {
-            this.consumerId = consumerId;
-            return this;
-        }
-
-        /**
-         * Setup custom Consumer Group - Default to OpenDCAE-DMaaPSub-ConsumerID
-         *
-         * @param consumerGroup - custom Consumer Group
-         * @return Builder object itself for chaining
-         */
-        public Builder setConsumerGroup(@Nonnull String consumerGroup) {
-            this.consumerGroup = consumerGroup;
-            return this;
-        }
-
-        /**
-         * Setup Custom Subscriber timeout in ms - Default to no timeout limit
-         *
-         * @param timeoutMS timeout in milliseconds
-         * @return Builder object itself for chaining
-         */
-        public Builder setTimeoutMS(@Nonnull Integer timeoutMS) {
-            this.timeoutMS = timeoutMS;
-            return this;
-        }
-
-        /**
-         * Setup custom Subscriber Message Limit - Default to no limit
-         *
-         * @param messageLimit message Limit
-         * @return Builder object itself for chaining
-         */
-        public Builder setMessageLimit(@Nonnull Integer messageLimit) {
-            this.messageLimit = messageLimit;
-            return this;
-        }
-
-        /**
-         * Builds Immutable instance of {@link DMaaPMRSubscriberConfig}
-         *
-         * @return immutable DMaaP Subscriber Config Object
-         */
-        public DMaaPMRSubscriberConfig build() {
-            return new DMaaPMRSubscriberConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
-                    contentType, consumerId, consumerGroup, timeoutMS, messageLimit);
-        }
-
-    }
-
-
-    /**
-     * DMaaP MR Subscriber Consumer Id
-     *
-     * @return consumer Id
-     */
-    public String getConsumerId() {
-        return consumerId;
-    }
-
-    /**
-     * DMaaP MR Subscriber Consumer Group
-     *
-     * @return consumer group
-     */
-    public String getConsumerGroup() {
-        return consumerGroup;
-    }
-
-    /**
-     * DMaaP MR Subscriber Timeout in ms
-     *
-     * @return subscriber timeout ms
-     */
-    public Integer getTimeoutMS() {
-        return timeoutMS;
-    }
-
-    /**
-     * DMaaP MR Subscriber message limit
-     *
-     * @return subscriber message limit
-     */
-    public Integer getMessageLimit() {
-        return messageLimit;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        if (!super.equals(o)) {
-            return false;
-        }
-        DMaaPMRSubscriberConfig that = (DMaaPMRSubscriberConfig) o;
-        return Objects.equal(consumerId, that.consumerId) &&
-                Objects.equal(consumerGroup, that.consumerGroup) &&
-                Objects.equal(timeoutMS, that.timeoutMS) &&
-                Objects.equal(messageLimit, that.messageLimit);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hashCode(super.hashCode(), consumerId, consumerGroup, timeoutMS, messageLimit);
-    }
-
-
-    @Override
-    public String toString() {
-        return Objects.toStringHelper(this)
-                .add("baseConfig", super.toString())
-                .add("consumerId", consumerId)
-                .add("consumerGroup", consumerGroup)
-                .add("timeoutMS", timeoutMS)
-                .add("messageLimit", messageLimit)
-                .toString();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.config;
+
+import com.google.common.base.Objects;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+
+import java.util.UUID;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ *      Immutable DMaaP MR Configuration for Subscriber.
+ * <p>
+ *      Use {@link DMaaPMRSubscriberConfig.Builder} to construct Subscriber Configuration
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/12/2016.
+ */
+public final class DMaaPMRSubscriberConfig extends DMaaPMRBaseConfig {
+
+    private final String consumerId;
+    private final String consumerGroup;
+    private final Integer timeoutMS;
+    private final Integer messageLimit;
+
+    private DMaaPMRSubscriberConfig(@Nonnull String hostName,
+                                    @Nonnull Integer portNumber,
+                                    @Nonnull String topicName,
+                                    @Nonnull String protocol,
+                                    String userName,
+                                    String userPassword,
+                                    @Nonnull String contentType,
+                                    @Nonnull String consumerId,
+                                    @Nonnull String consumerGroup,
+                                    @Nonnull Integer timeoutMS,
+                                    @Nonnull Integer messageLimit) {
+        this.hostName = hostName;
+        this.portNumber = portNumber;
+        this.topicName = topicName;
+        this.protocol = protocol;
+        this.userName = userName;
+        this.userPassword = userPassword;
+        this.contentType = contentType;
+        this.consumerId = consumerId;
+        this.consumerGroup = consumerGroup;
+        this.timeoutMS = timeoutMS;
+        this.messageLimit = messageLimit;
+    }
+
+    /**
+     * Builder to initialize immutable {@link DMaaPMRSubscriberConfig} object
+     */
+    public static class Builder {
+
+        private String hostName;
+        private Integer portNumber;
+        private String topicName;
+        private String userName;
+        private String userPassword;
+        private String protocol;
+        private String contentType;
+        private String consumerId;
+        private String consumerGroup;
+        private Integer timeoutMS;
+        private Integer messageLimit;
+
+        public Builder(@Nonnull String hostName,
+                       @Nonnull String topicName) {
+            // Required Values
+            this.hostName = hostName;
+            this.topicName = topicName;
+
+            // Default values
+            this.portNumber = AnalyticsConstants.DEFAULT_PORT_NUMBER;
+            this.userName = AnalyticsConstants.DEFAULT_USER_NAME;
+            this.userPassword = AnalyticsConstants.DEFAULT_USER_PASSWORD;
+            this.protocol = AnalyticsConstants.DEFAULT_PROTOCOL;
+            this.contentType = AnalyticsConstants.DEFAULT_CONTENT_TYPE;
+            this.consumerId = UUID.randomUUID().toString(); // consumer is assigned a random id by default
+            this.consumerGroup = AnalyticsConstants.DEFAULT_SUBSCRIBER_GROUP_PREFIX + consumerId; // random group
+            this.timeoutMS = AnalyticsConstants.DEFAULT_SUBSCRIBER_TIMEOUT_MS; // defaults to 10ms timeout
+            this.messageLimit = AnalyticsConstants.DEFAULT_SUBSCRIBER_MESSAGE_LIMIT; // defaults to 1000 message limit
+        }
+
+
+        /**
+         * Setup for custom host port number - Defaults to 80.
+         *
+         * @param portNumber custom port number
+         * @return Builder object itself for chaining
+         */
+        public Builder setPortNumber(@Nonnull Integer portNumber) {
+            this.portNumber = portNumber;
+            return this;
+        }
+
+
+        /**
+         * Setup user name for authentication. If no username is provided authentication will be disabled
+         *
+         * @param userName user name for DMaaP Topic Authentication
+         * @return Builder object itself for chaining
+         */
+        public Builder setUserName(@Nonnull String userName) {
+            this.userName = userName;
+            return this;
+        }
+
+
+        /**
+         * Setup user password for authentication. If no password is provided authentication will be disabled
+         *
+         * @param userPassword user password for DMaaP Topic Authentication
+         * @return Builder object itself for chaining
+         */
+        public Builder setUserPassword(@Nonnull String userPassword) {
+            this.userPassword = userPassword;
+            return this;
+        }
+
+
+        /**
+         * Setup custom Subscriber protocol - Defaults to https.
+         * Note: Only http and https are currently supported.
+         *
+         * @param protocol protocol e.g. https or http
+         * @return Builder object itself for chaining
+         */
+        public Builder setProtocol(@Nonnull String protocol) {
+
+            this.protocol = normalizeValidateProtocol(protocol);
+            return this;
+        }
+
+        /**
+         * Setup custom Subscriber content-type - Defaults to application/json
+         *
+         * @param contentType content type e.g. application/json
+         * @return Builder object itself for chaining
+         */
+        public Builder setContentType(@Nonnull String contentType) {
+            final String normalizedContentType = normalizeValidateContentType(contentType);
+            this.contentType = normalizedContentType;
+            return this;
+        }
+
+
+        /**
+         * Setup custom Consumer Id - Defaults to random Id
+         *
+         * @param consumerId - custom consumer ID
+         * @return Builder object itself for chaining
+         */
+        public Builder setConsumerId(@Nonnull String consumerId) {
+            this.consumerId = consumerId;
+            return this;
+        }
+
+        /**
+         * Setup custom Consumer Group - Default to OpenDCAE-DMaaPSub-ConsumerID
+         *
+         * @param consumerGroup - custom Consumer Group
+         * @return Builder object itself for chaining
+         */
+        public Builder setConsumerGroup(@Nonnull String consumerGroup) {
+            this.consumerGroup = consumerGroup;
+            return this;
+        }
+
+        /**
+         * Setup Custom Subscriber timeout in ms - Default to no timeout limit
+         *
+         * @param timeoutMS timeout in milliseconds
+         * @return Builder object itself for chaining
+         */
+        public Builder setTimeoutMS(@Nonnull Integer timeoutMS) {
+            this.timeoutMS = timeoutMS;
+            return this;
+        }
+
+        /**
+         * Setup custom Subscriber Message Limit - Default to no limit
+         *
+         * @param messageLimit message Limit
+         * @return Builder object itself for chaining
+         */
+        public Builder setMessageLimit(@Nonnull Integer messageLimit) {
+            this.messageLimit = messageLimit;
+            return this;
+        }
+
+        /**
+         * Builds Immutable instance of {@link DMaaPMRSubscriberConfig}
+         *
+         * @return immutable DMaaP Subscriber Config Object
+         */
+        public DMaaPMRSubscriberConfig build() {
+            return new DMaaPMRSubscriberConfig(hostName, portNumber, topicName, protocol, userName, userPassword,
+                    contentType, consumerId, consumerGroup, timeoutMS, messageLimit);
+        }
+
+    }
+
+
+    /**
+     * DMaaP MR Subscriber Consumer Id
+     *
+     * @return consumer Id
+     */
+    public String getConsumerId() {
+        return consumerId;
+    }
+
+    /**
+     * DMaaP MR Subscriber Consumer Group
+     *
+     * @return consumer group
+     */
+    public String getConsumerGroup() {
+        return consumerGroup;
+    }
+
+    /**
+     * DMaaP MR Subscriber Timeout in ms
+     *
+     * @return subscriber timeout ms
+     */
+    public Integer getTimeoutMS() {
+        return timeoutMS;
+    }
+
+    /**
+     * DMaaP MR Subscriber message limit
+     *
+     * @return subscriber message limit
+     */
+    public Integer getMessageLimit() {
+        return messageLimit;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        DMaaPMRSubscriberConfig that = (DMaaPMRSubscriberConfig) o;
+        return Objects.equal(consumerId, that.consumerId) &&
+                Objects.equal(consumerGroup, that.consumerGroup) &&
+                Objects.equal(timeoutMS, that.timeoutMS) &&
+                Objects.equal(messageLimit, that.messageLimit);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(super.hashCode(), consumerId, consumerGroup, timeoutMS, messageLimit);
+    }
+
+
+    @Override
+    public String toString() {
+        return Objects.toStringHelper(this)
+                .add("baseConfig", super.toString())
+                .add("consumerId", consumerId)
+                .add("consumerGroup", consumerGroup)
+                .add("timeoutMS", timeoutMS)
+                .add("messageLimit", messageLimit)
+                .toString();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
index 845f6df..b4a29a1 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponse.java
@@ -1,38 +1,38 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-/**
- * <p>
- *      Contract for all DMaaPMR Publisher Response
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRPublisherResponse extends DMaaPMRResponse {
-
-
-    /**
-     * Gets number of pending messages
-     *
-     * @return pending messages in the batch queue
-     */
-    int getPendingMessagesCount();
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.response;
+
+/**
+ * <p>
+ *      Contract for all DMaaPMR Publisher Response
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRPublisherResponse extends DMaaPMRResponse {
+
+
+    /**
+     * Gets number of pending messages
+     *
+     * @return pending messages in the batch queue
+     */
+    int getPendingMessagesCount();
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
index 92996b4..879feac 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRPublisherResponseImpl.java
@@ -1,70 +1,70 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import com.google.common.base.Objects;
-
-import javax.annotation.Nonnull;
-
-/**
- * <p>
- *      An simple implementation of {@link DMaaPMRPublisherResponse}
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRPublisherResponseImpl implements DMaaPMRPublisherResponse {
-
-    private final Integer responseCode;
-    private final String responseMessage;
-    private final int pendingMessagesCount;
-
-    public DMaaPMRPublisherResponseImpl(@Nonnull Integer responseCode,
-                                        @Nonnull String responseMessage,
-                                        int pendingMessagesCount) {
-        this.responseCode = responseCode;
-        this.responseMessage = responseMessage;
-        this.pendingMessagesCount = pendingMessagesCount;
-    }
-
-    @Override
-    public Integer getResponseCode() {
-        return responseCode;
-    }
-
-    @Override
-    public String getResponseMessage() {
-        return responseMessage;
-    }
-
-    @Override
-    public int getPendingMessagesCount() {
-        return pendingMessagesCount;
-    }
-
-    @Override
-    public String toString() {
-        return Objects.toStringHelper(this)
-                .add("responseCode", responseCode)
-                .add("responseMessage", responseMessage)
-                .add("pendingMessagesCount", pendingMessagesCount)
-                .toString();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.response;
+
+import com.google.common.base.Objects;
+
+import javax.annotation.Nonnull;
+
+/**
+ * <p>
+ *      An simple implementation of {@link DMaaPMRPublisherResponse}
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRPublisherResponseImpl implements DMaaPMRPublisherResponse {
+
+    private final Integer responseCode;
+    private final String responseMessage;
+    private final int pendingMessagesCount;
+
+    public DMaaPMRPublisherResponseImpl(@Nonnull Integer responseCode,
+                                        @Nonnull String responseMessage,
+                                        int pendingMessagesCount) {
+        this.responseCode = responseCode;
+        this.responseMessage = responseMessage;
+        this.pendingMessagesCount = pendingMessagesCount;
+    }
+
+    @Override
+    public Integer getResponseCode() {
+        return responseCode;
+    }
+
+    @Override
+    public String getResponseMessage() {
+        return responseMessage;
+    }
+
+    @Override
+    public int getPendingMessagesCount() {
+        return pendingMessagesCount;
+    }
+
+    @Override
+    public String toString() {
+        return Objects.toStringHelper(this)
+                .add("responseCode", responseCode)
+                .add("responseMessage", responseMessage)
+                .add("pendingMessagesCount", pendingMessagesCount)
+                .toString();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
index 661a7e5..4ac5cf8 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRResponse.java
@@ -1,46 +1,46 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-/**
- * <p>
- *      Contract for DMaaP MR Responses
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRResponse {
-
-    /**
-     * Gets HTTP Response Code
-     *
-     * @return HTTP Response code as String
-     */
-    Integer getResponseCode();
-
-    /**
-     * Gets Response Message
-     *
-     * @return Response Message
-     */
-    String getResponseMessage();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.response;
+
+/**
+ * <p>
+ *      Contract for DMaaP MR Responses
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRResponse {
+
+    /**
+     * Gets HTTP Response Code
+     *
+     * @return HTTP Response code as String
+     */
+    Integer getResponseCode();
+
+    /**
+     * Gets Response Message
+     *
+     * @return Response Message
+     */
+    String getResponseMessage();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
index 228e2d3..5e90e24 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponse.java
@@ -1,40 +1,40 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import java.util.List;
-
-/**
- * <p>
- *     Contract for all DMaaP MR Subscriber Responses
- * </p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRSubscriberResponse extends DMaaPMRResponse {
-
-    /**
-     * Returns message fetched from DMaaP MR Topic
-     *
-     * @return collection of actual message retrieved from DMaaP MR Topic
-     */
-    List<String> getFetchedMessages();
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.response;
+
+import java.util.List;
+
+/**
+ * <p>
+ *     Contract for all DMaaP MR Subscriber Responses
+ * </p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRSubscriberResponse extends DMaaPMRResponse {
+
+    /**
+     * Returns message fetched from DMaaP MR Topic
+     *
+     * @return collection of actual message retrieved from DMaaP MR Topic
+     */
+    List<String> getFetchedMessages();
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
index 08f0dab..160491c 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/domain/response/DMaaPMRSubscriberResponseImpl.java
@@ -1,80 +1,80 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.domain.response;
-
-import com.google.common.base.Objects;
-import com.google.common.collect.ImmutableList;
-
-import java.util.List;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import static java.util.Collections.unmodifiableList;
-
-/**
- * <p>
- *      A simple implementation for {@link DMaaPMRSubscriberResponse}
- * <p>
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRSubscriberResponseImpl implements DMaaPMRSubscriberResponse {
-
-    private final Integer responseCode;
-    private final String responseMessage;
-    private final List<String> fetchedMessages;
-
-    public DMaaPMRSubscriberResponseImpl(@Nonnull Integer responseCode,
-                                         @Nonnull String responseMessage,
-                                         @Nullable List<String> fetchedMessages) {
-        this.responseCode = responseCode;
-        this.responseMessage = responseMessage;
-        this.fetchedMessages = fetchedMessages != null ? fetchedMessages : ImmutableList.<String>of();
-    }
-
-    public DMaaPMRSubscriberResponseImpl(Integer responseCode, String responseMessage) {
-        this(responseCode, responseMessage, null);
-    }
-
-    @Override
-    public Integer getResponseCode() {
-        return responseCode;
-    }
-
-    @Override
-    public String getResponseMessage() {
-        return responseMessage;
-    }
-
-    @Override
-    public List<String> getFetchedMessages() {
-        return unmodifiableList(fetchedMessages);
-    }
-
-    @Override
-    public String toString() {
-        return Objects.toStringHelper(this)
-                .add("responseCode", responseCode)
-                .add("responseMessage", responseMessage)
-                .add("fetchedMessages(size)", fetchedMessages.size())
-                .toString();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.domain.response;
+
+import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static java.util.Collections.unmodifiableList;
+
+/**
+ * <p>
+ *      A simple implementation for {@link DMaaPMRSubscriberResponse}
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRSubscriberResponseImpl implements DMaaPMRSubscriberResponse {
+
+    private final Integer responseCode;
+    private final String responseMessage;
+    private final List<String> fetchedMessages;
+
+    public DMaaPMRSubscriberResponseImpl(@Nonnull Integer responseCode,
+                                         @Nonnull String responseMessage,
+                                         @Nullable List<String> fetchedMessages) {
+        this.responseCode = responseCode;
+        this.responseMessage = responseMessage;
+        this.fetchedMessages = fetchedMessages != null ? fetchedMessages : ImmutableList.<String>of();
+    }
+
+    public DMaaPMRSubscriberResponseImpl(Integer responseCode, String responseMessage) {
+        this(responseCode, responseMessage, null);
+    }
+
+    @Override
+    public Integer getResponseCode() {
+        return responseCode;
+    }
+
+    @Override
+    public String getResponseMessage() {
+        return responseMessage;
+    }
+
+    @Override
+    public List<String> getFetchedMessages() {
+        return unmodifiableList(fetchedMessages);
+    }
+
+    @Override
+    public String toString() {
+        return Objects.toStringHelper(this)
+                .add("responseCode", responseCode)
+                .add("responseMessage", responseMessage)
+                .add("fetchedMessages(size)", fetchedMessages.size())
+                .toString();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
index e85b377..ab08642 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/module/AnalyticsDMaaPModule.java
@@ -1,62 +1,62 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.module;
-
-import com.google.inject.AbstractModule;
-import com.google.inject.assistedinject.FactoryModuleBuilder;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberImpl;
-
-/**
- * Guice Module to wire concrete implementations with interfaces
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public class AnalyticsDMaaPModule extends AbstractModule {
-
-
-    @Override
-    protected void configure() {
-
-        // Bind Http Client
-        bind(CloseableHttpClient.class).toInstance(HttpClients.createDefault());
-
-        // Bind Publishing queue
-        install(new FactoryModuleBuilder().implement(DMaaPMRPublisherQueue.class, DMaaPMRPublisherQueueImpl.class)
-                .build(DMaaPMRPublisherQueueFactory.class));
-
-        install(new FactoryModuleBuilder().implement(DMaaPMRPublisher.class, DMaaPMRPublisherImpl.class)
-                .build(DMaaPMRPublisherFactory.class));
-
-        install(new FactoryModuleBuilder().implement(DMaaPMRSubscriber.class, DMaaPMRSubscriberImpl.class)
-                .build(DMaaPMRSubscriberFactory.class));
-
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.module;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueueImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberFactory;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriberImpl;
+
+/**
+ * Guice Module to wire concrete implementations with interfaces
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public class AnalyticsDMaaPModule extends AbstractModule {
+
+
+    @Override
+    protected void configure() {
+
+        // Bind Http Client
+        bind(CloseableHttpClient.class).toInstance(HttpClients.createDefault());
+
+        // Bind Publishing queue
+        install(new FactoryModuleBuilder().implement(DMaaPMRPublisherQueue.class, DMaaPMRPublisherQueueImpl.class)
+                .build(DMaaPMRPublisherQueueFactory.class));
+
+        install(new FactoryModuleBuilder().implement(DMaaPMRPublisher.class, DMaaPMRPublisherImpl.class)
+                .build(DMaaPMRPublisherFactory.class));
+
+        install(new FactoryModuleBuilder().implement(DMaaPMRSubscriber.class, DMaaPMRSubscriberImpl.class)
+                .build(DMaaPMRSubscriberFactory.class));
+
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
index 22f25d1..f381b5f 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/BaseDMaaPMRComponent.java
@@ -1,378 +1,378 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Optional;
-import org.apache.commons.codec.binary.Base64;
-import org.apache.commons.lang3.StringEscapeUtils;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.ResponseHandler;
-import org.apache.http.client.utils.URIBuilder;
-import org.apache.http.util.EntityUtils;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponseImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponseImpl;
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.charset.Charset;
-import java.util.LinkedList;
-import java.util.List;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-
-import static java.lang.String.format;
-
-/**
- * Base class for DMaaP MR Publishers and Subscriber Implementations containing various utility methods
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public abstract class BaseDMaaPMRComponent implements DMaaPMRComponent {
-
-    private static final Logger LOG = LoggerFactory.getLogger(BaseDMaaPMRComponent.class);
-
-    private static final ObjectMapper objectMapper = new ObjectMapper();
-
-    /**
-     * Creates Base64 encoded Auth Header for given userName and Password
-     * If either user name of password are null return absent
-     *
-     * @param userName username
-     * @param userPassword user password
-     * @return base64 encoded auth header if username or password are both non null
-     */
-    protected static Optional<String> getAuthHeader(@Nullable final String userName,
-                                                    @Nullable final String userPassword) {
-        if (userName == null || userPassword == null) {
-            return Optional.absent();
-        } else {
-            final String auth = userName + ":" + userPassword;
-            final Charset isoCharset = Charset.forName("ISO-8859-1");
-            byte[] encodedAuth = Base64.encodeBase64(auth.getBytes(isoCharset));
-            return Optional.of("Basic " + new String(encodedAuth, isoCharset));
-        }
-    }
-
-
-    /**
-     * Creates Publisher URI for given {@link DMaaPMRPublisherConfig}
-     *
-     * @param publisherConfig publisher settings
-     *
-     * @return DMaaP MR Publisher Topic URI that can be used to post messages to MR Topic
-     */
-    protected static URI createPublisherURI(final DMaaPMRPublisherConfig publisherConfig) {
-        final String hostName = publisherConfig.getHostName();
-        final Integer portNumber = publisherConfig.getPortNumber();
-        final String getProtocol = publisherConfig.getProtocol();
-        final String topicName = publisherConfig.getTopicName();
-        URI publisherURI = null;
-        try {
-            publisherURI = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
-                    .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX + topicName).build();
-        } catch (URISyntaxException e) {
-            final String errorMessage = format("Error while creating publisher URI: %s", e);
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-        LOG.info("Created DMaaP MR Publisher URI: {}", publisherURI);
-        return publisherURI;
-    }
-
-
-    /**
-     * Creates Subscriber URI for given {@link DMaaPMRSubscriberConfig}
-     *
-     * @param subscriberConfig subscriber settings
-     *
-     * @return DMaaP MR Subscriber Topic URI that can be used to fetch messages from MR topic
-     */
-    protected static URI createSubscriberURI(final DMaaPMRSubscriberConfig subscriberConfig) {
-        final String hostName = subscriberConfig.getHostName();
-        final Integer portNumber = subscriberConfig.getPortNumber();
-        final String getProtocol = subscriberConfig.getProtocol();
-        final String topicName = subscriberConfig.getTopicName();
-        final String consumerId = subscriberConfig.getConsumerId();
-        final String consumerGroup = subscriberConfig.getConsumerGroup();
-        final Integer timeoutMS = subscriberConfig.getTimeoutMS();
-        final Integer messageLimit = subscriberConfig.getMessageLimit();
-        URI subscriberURI = null;
-        try {
-            URIBuilder uriBuilder = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
-                    .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX
-                            + topicName + "/"
-                            + consumerGroup + "/" +
-                            consumerId);
-            // add query params if present
-            if (timeoutMS > 0) {
-                uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME, timeoutMS.toString());
-            }
-            if (messageLimit > 0) {
-                uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME,
-                        messageLimit.toString());
-            }
-            subscriberURI = uriBuilder.build();
-
-        } catch (URISyntaxException e) {
-            final String errorMessage = format("Error while creating subscriber URI: %s", e);
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-
-        LOG.info("Created DMaaP MR Subscriber URI: {}", subscriberURI);
-        return subscriberURI;
-    }
-
-
-    /**
-     *  Creates 202 (Accepted) Response code message
-     *
-     * @param batchQueueSize batch Queue size
-     *
-     * @return response with 202 message code
-     */
-    protected static DMaaPMRPublisherResponse createPublisherAcceptedResponse(int batchQueueSize) {
-        return createPublisherResponse(HTTPUtils.HTTP_ACCEPTED_RESPONSE_CODE,
-                "Accepted - Messages queued for batch publishing to MR Topic", batchQueueSize);
-    }
-
-
-    /**
-     *  Creates 204 (No Content) Response code message
-     *
-     * @return response with 204 message code
-     */
-    protected static DMaaPMRPublisherResponse createPublisherNoContentResponse() {
-        return createPublisherResponse(HTTPUtils.HTTP_NO_CONTENT_RESPONSE_CODE,
-                "No Content - No Messages in batch queue for flushing to MR Topic", 0);
-    }
-
-
-    /**
-     * Creates Publisher Response for given response code, response Message and pending Message Count
-     *
-     * @param responseCode HTTP Status Code
-     * @param responseMessage response message
-     * @param pendingMessages pending messages in batch queue
-     *
-     * @return DMaaP MR Publisher Response
-     */
-    protected static DMaaPMRPublisherResponse createPublisherResponse(int responseCode, String
-            responseMessage, int pendingMessages) {
-        return new DMaaPMRPublisherResponseImpl(responseCode, responseMessage, pendingMessages);
-    }
-
-
-    /**
-     * Returns weekly consistent pending messages in batch queue
-     *
-     * @param publisherQueue batch queue
-     * @param publisherConfig publisher settings
-     *
-     * @return pending messages to be published
-     */
-    protected static int getPendingMessages(@Nonnull final DMaaPMRPublisherQueue publisherQueue,
-                                            @Nonnull final DMaaPMRPublisherConfig publisherConfig) {
-        return publisherConfig.getMaxBatchSize() - publisherQueue.getBatchQueueRemainingSize();
-    }
-
-
-    /**
-     * Creates Subscriber Response for give response Code, response Message and fetch messages
-     *
-     * @param responseCode response Code
-     * @param responseMessage response Message
-     * @param fetchedMessages fetched messages
-     *
-     * @return DMaaP MR Subscriber Response
-     */
-    protected static DMaaPMRSubscriberResponse createSubscriberResponse(int responseCode, String
-            responseMessage, List<String> fetchedMessages) {
-        if (fetchedMessages == null) {
-            return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage);
-        } else {
-            return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage, fetchedMessages);
-        }
-    }
-
-
-    /**
-     * Custom response handler which extract status code and response body
-     *
-     * @return Pair containing Response code and response body
-     */
-    protected static ResponseHandler<Pair<Integer, String>> responseHandler() {
-        return new ResponseHandler<Pair<Integer, String>>() {
-            @Override
-            public Pair<Integer, String> handleResponse(HttpResponse response) throws IOException {
-                // Get Response status code
-                final int status = response.getStatusLine().getStatusCode();
-                final HttpEntity responseEntity = response.getEntity();
-                // If response entity is not null - extract response body as string
-                String responseEntityString = "";
-                if (responseEntity != null) {
-                    responseEntityString = EntityUtils.toString(responseEntity);
-                }
-                return new ImmutablePair<>(status, responseEntityString);
-            }
-        };
-    }
-
-
-    /**
-     *  Adds message to Publisher recovery queue. If recovery queue is full throws an error as messages will
-     *  be lost
-     *
-     * @param publisherQueue publisher queue
-     * @param messages recoverable messages to be published to recovery queue
-     */
-    protected static void addMessagesToRecoveryQueue(DMaaPMRPublisherQueue publisherQueue,
-                                                     List<String> messages) {
-        try {
-            publisherQueue.addRecoverableMessages(messages);
-
-            LOG.debug("Messages Added to Recovery Queue. Messages Size: {}, Recovery Queue Remaining Size: {}",
-                    messages.size(), publisherQueue.getBatchQueueRemainingSize());
-
-        } catch (IllegalStateException e) {
-            final String errorMessage = format("Unable to put messages in recovery queue. Messages will be lost. " +
-                            "Recovery Queue might be full. Message Size: %d, Recovery Queue Remaining Capacity: %d",
-                    messages.size(), publisherQueue.getRecoveryQueueRemainingSize());
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-    }
-
-
-    /**
-     * Converts List of messages to Json String Array which can be published to DMaaP MR topic.
-     *
-     * @param messages messages that need to parsed to Json Array representation
-     * @return json string representation of message
-     */
-    protected static String convertToJsonString(@Nullable final List<String> messages) {
-        // If messages are null or empty just return empty array
-        if (messages == null || messages.isEmpty()) {
-            return "[]";
-        }
-
-
-        List<JsonNode> jsonMessageObjectsList = new LinkedList<>();
-
-        try {
-            for (String message : messages) {
-                final JsonNode jsonNode = objectMapper.readTree(message);
-                jsonMessageObjectsList.add(jsonNode);
-            }
-            return objectMapper.writeValueAsString(jsonMessageObjectsList);
-        } catch (JsonProcessingException e) {
-            final String errorMessage =
-                    format("Unable to convert publisher messages to Json. Messages: %s, Json Error: %s",
-                            messages, e);
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-
-        } catch (IOException e) {
-            final String errorMessage =
-                    format("IO Exception while converting publisher messages to Json. Messages: %s, Json Error: %s",
-                            messages, e);
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-    }
-
-
-    /**
-     * Converts subscriber messages json string to List of messages. If message Json String is empty
-     * or null
-     *
-     * @param messagesJsonString json messages String
-     *
-     * @return List containing DMaaP MR Messages
-     */
-    protected static List<String> convertJsonToStringMessages(@Nullable final String messagesJsonString) {
-
-        final LinkedList<String> messages = new LinkedList<>();
-
-        // If message string is not null or not empty parse json message array to List of string messages
-        if (messagesJsonString != null && !messagesJsonString.trim().isEmpty()
-                && !("[]").equals(messagesJsonString.trim())) {
-
-            try {
-                // get root node
-                final JsonNode rootNode = objectMapper.readTree(messagesJsonString);
-                // iterate over root node and parse arrays messages
-                for (JsonNode jsonNode : rootNode) {
-                    // if array parse it is array of messages
-                    final String incomingMessageString = jsonNode.toString();
-                    if (jsonNode.isArray()) {
-                        final List messageList = objectMapper.readValue(incomingMessageString, List.class);
-                        for (Object message : messageList) {
-                            final String jsonMessageString = objectMapper.writeValueAsString(message);
-                            addUnescapedJsonToMessage(messages, jsonMessageString);
-                        }
-                    } else {
-                        // parse it as object
-                        addUnescapedJsonToMessage(messages, incomingMessageString);
-                    }
-                }
-
-            } catch (IOException e) {
-                final String errorMessage =
-                        format("Unable to convert subscriber Json String to Messages. Subscriber Response String: %s," +
-                                " Json Error: %s", messagesJsonString, e);
-                throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-            }
-
-        }
-        return messages;
-    }
-
-    /**
-     * Adds unescaped Json messages to given messages list
-     *
-     * @param messages message list in which unescaped messages will be added
-     * @param incomingMessageString incoming message string that may need to be escaped
-     */
-    private static void addUnescapedJsonToMessage(List<String> messages, String incomingMessageString) {
-        if (incomingMessageString.startsWith("\"") && incomingMessageString.endsWith("\"")) {
-            messages.add(StringEscapeUtils.unescapeJson(
-                    incomingMessageString.substring(1, incomingMessageString.length() - 1)));
-        } else {
-            messages.add(StringEscapeUtils.unescapeJson(incomingMessageString));
-        }
-    }
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang3.StringEscapeUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.util.EntityUtils;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.common.utils.HTTPUtils;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponseImpl;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponseImpl;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisherQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.Charset;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static java.lang.String.format;
+
+/**
+ * Base class for DMaaP MR Publishers and Subscriber Implementations containing various utility methods
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public abstract class BaseDMaaPMRComponent implements DMaaPMRComponent {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BaseDMaaPMRComponent.class);
+
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
+    /**
+     * Creates Base64 encoded Auth Header for given userName and Password
+     * If either user name of password are null return absent
+     *
+     * @param userName username
+     * @param userPassword user password
+     * @return base64 encoded auth header if username or password are both non null
+     */
+    protected static Optional<String> getAuthHeader(@Nullable final String userName,
+                                                    @Nullable final String userPassword) {
+        if (userName == null || userPassword == null) {
+            return Optional.absent();
+        } else {
+            final String auth = userName + ":" + userPassword;
+            final Charset isoCharset = Charset.forName("ISO-8859-1");
+            byte[] encodedAuth = Base64.encodeBase64(auth.getBytes(isoCharset));
+            return Optional.of("Basic " + new String(encodedAuth, isoCharset));
+        }
+    }
+
+
+    /**
+     * Creates Publisher URI for given {@link DMaaPMRPublisherConfig}
+     *
+     * @param publisherConfig publisher settings
+     *
+     * @return DMaaP MR Publisher Topic URI that can be used to post messages to MR Topic
+     */
+    protected static URI createPublisherURI(final DMaaPMRPublisherConfig publisherConfig) {
+        final String hostName = publisherConfig.getHostName();
+        final Integer portNumber = publisherConfig.getPortNumber();
+        final String getProtocol = publisherConfig.getProtocol();
+        final String topicName = publisherConfig.getTopicName();
+        URI publisherURI = null;
+        try {
+            publisherURI = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
+                    .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX + topicName).build();
+        } catch (URISyntaxException e) {
+            final String errorMessage = format("Error while creating publisher URI: %s", e);
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+        LOG.info("Created DMaaP MR Publisher URI: {}", publisherURI);
+        return publisherURI;
+    }
+
+
+    /**
+     * Creates Subscriber URI for given {@link DMaaPMRSubscriberConfig}
+     *
+     * @param subscriberConfig subscriber settings
+     *
+     * @return DMaaP MR Subscriber Topic URI that can be used to fetch messages from MR topic
+     */
+    protected static URI createSubscriberURI(final DMaaPMRSubscriberConfig subscriberConfig) {
+        final String hostName = subscriberConfig.getHostName();
+        final Integer portNumber = subscriberConfig.getPortNumber();
+        final String getProtocol = subscriberConfig.getProtocol();
+        final String topicName = subscriberConfig.getTopicName();
+        final String consumerId = subscriberConfig.getConsumerId();
+        final String consumerGroup = subscriberConfig.getConsumerGroup();
+        final Integer timeoutMS = subscriberConfig.getTimeoutMS();
+        final Integer messageLimit = subscriberConfig.getMessageLimit();
+        URI subscriberURI = null;
+        try {
+            URIBuilder uriBuilder = new URIBuilder().setScheme(getProtocol).setHost(hostName).setPort(portNumber)
+                    .setPath(AnalyticsConstants.DMAAP_URI_PATH_PREFIX
+                            + topicName + "/"
+                            + consumerGroup + "/" +
+                            consumerId);
+            // add query params if present
+            if (timeoutMS > 0) {
+                uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_TIMEOUT_QUERY_PARAM_NAME, timeoutMS.toString());
+            }
+            if (messageLimit > 0) {
+                uriBuilder.addParameter(AnalyticsConstants.SUBSCRIBER_MSG_LIMIT_QUERY_PARAM_NAME,
+                        messageLimit.toString());
+            }
+            subscriberURI = uriBuilder.build();
+
+        } catch (URISyntaxException e) {
+            final String errorMessage = format("Error while creating subscriber URI: %s", e);
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+
+        LOG.info("Created DMaaP MR Subscriber URI: {}", subscriberURI);
+        return subscriberURI;
+    }
+
+
+    /**
+     *  Creates 202 (Accepted) Response code message
+     *
+     * @param batchQueueSize batch Queue size
+     *
+     * @return response with 202 message code
+     */
+    protected static DMaaPMRPublisherResponse createPublisherAcceptedResponse(int batchQueueSize) {
+        return createPublisherResponse(HTTPUtils.HTTP_ACCEPTED_RESPONSE_CODE,
+                "Accepted - Messages queued for batch publishing to MR Topic", batchQueueSize);
+    }
+
+
+    /**
+     *  Creates 204 (No Content) Response code message
+     *
+     * @return response with 204 message code
+     */
+    protected static DMaaPMRPublisherResponse createPublisherNoContentResponse() {
+        return createPublisherResponse(HTTPUtils.HTTP_NO_CONTENT_RESPONSE_CODE,
+                "No Content - No Messages in batch queue for flushing to MR Topic", 0);
+    }
+
+
+    /**
+     * Creates Publisher Response for given response code, response Message and pending Message Count
+     *
+     * @param responseCode HTTP Status Code
+     * @param responseMessage response message
+     * @param pendingMessages pending messages in batch queue
+     *
+     * @return DMaaP MR Publisher Response
+     */
+    protected static DMaaPMRPublisherResponse createPublisherResponse(int responseCode, String
+            responseMessage, int pendingMessages) {
+        return new DMaaPMRPublisherResponseImpl(responseCode, responseMessage, pendingMessages);
+    }
+
+
+    /**
+     * Returns weekly consistent pending messages in batch queue
+     *
+     * @param publisherQueue batch queue
+     * @param publisherConfig publisher settings
+     *
+     * @return pending messages to be published
+     */
+    protected static int getPendingMessages(@Nonnull final DMaaPMRPublisherQueue publisherQueue,
+                                            @Nonnull final DMaaPMRPublisherConfig publisherConfig) {
+        return publisherConfig.getMaxBatchSize() - publisherQueue.getBatchQueueRemainingSize();
+    }
+
+
+    /**
+     * Creates Subscriber Response for give response Code, response Message and fetch messages
+     *
+     * @param responseCode response Code
+     * @param responseMessage response Message
+     * @param fetchedMessages fetched messages
+     *
+     * @return DMaaP MR Subscriber Response
+     */
+    protected static DMaaPMRSubscriberResponse createSubscriberResponse(int responseCode, String
+            responseMessage, List<String> fetchedMessages) {
+        if (fetchedMessages == null) {
+            return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage);
+        } else {
+            return new DMaaPMRSubscriberResponseImpl(responseCode, responseMessage, fetchedMessages);
+        }
+    }
+
+
+    /**
+     * Custom response handler which extract status code and response body
+     *
+     * @return Pair containing Response code and response body
+     */
+    protected static ResponseHandler<Pair<Integer, String>> responseHandler() {
+        return new ResponseHandler<Pair<Integer, String>>() {
+            @Override
+            public Pair<Integer, String> handleResponse(HttpResponse response) throws IOException {
+                // Get Response status code
+                final int status = response.getStatusLine().getStatusCode();
+                final HttpEntity responseEntity = response.getEntity();
+                // If response entity is not null - extract response body as string
+                String responseEntityString = "";
+                if (responseEntity != null) {
+                    responseEntityString = EntityUtils.toString(responseEntity);
+                }
+                return new ImmutablePair<>(status, responseEntityString);
+            }
+        };
+    }
+
+
+    /**
+     *  Adds message to Publisher recovery queue. If recovery queue is full throws an error as messages will
+     *  be lost
+     *
+     * @param publisherQueue publisher queue
+     * @param messages recoverable messages to be published to recovery queue
+     */
+    protected static void addMessagesToRecoveryQueue(DMaaPMRPublisherQueue publisherQueue,
+                                                     List<String> messages) {
+        try {
+            publisherQueue.addRecoverableMessages(messages);
+
+            LOG.debug("Messages Added to Recovery Queue. Messages Size: {}, Recovery Queue Remaining Size: {}",
+                    messages.size(), publisherQueue.getBatchQueueRemainingSize());
+
+        } catch (IllegalStateException e) {
+            final String errorMessage = format("Unable to put messages in recovery queue. Messages will be lost. " +
+                            "Recovery Queue might be full. Message Size: %d, Recovery Queue Remaining Capacity: %d",
+                    messages.size(), publisherQueue.getRecoveryQueueRemainingSize());
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+    }
+
+
+    /**
+     * Converts List of messages to Json String Array which can be published to DMaaP MR topic.
+     *
+     * @param messages messages that need to parsed to Json Array representation
+     * @return json string representation of message
+     */
+    protected static String convertToJsonString(@Nullable final List<String> messages) {
+        // If messages are null or empty just return empty array
+        if (messages == null || messages.isEmpty()) {
+            return "[]";
+        }
+
+
+        List<JsonNode> jsonMessageObjectsList = new LinkedList<>();
+
+        try {
+            for (String message : messages) {
+                final JsonNode jsonNode = objectMapper.readTree(message);
+                jsonMessageObjectsList.add(jsonNode);
+            }
+            return objectMapper.writeValueAsString(jsonMessageObjectsList);
+        } catch (JsonProcessingException e) {
+            final String errorMessage =
+                    format("Unable to convert publisher messages to Json. Messages: %s, Json Error: %s",
+                            messages, e);
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+
+        } catch (IOException e) {
+            final String errorMessage =
+                    format("IO Exception while converting publisher messages to Json. Messages: %s, Json Error: %s",
+                            messages, e);
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+    }
+
+
+    /**
+     * Converts subscriber messages json string to List of messages. If message Json String is empty
+     * or null
+     *
+     * @param messagesJsonString json messages String
+     *
+     * @return List containing DMaaP MR Messages
+     */
+    protected static List<String> convertJsonToStringMessages(@Nullable final String messagesJsonString) {
+
+        final LinkedList<String> messages = new LinkedList<>();
+
+        // If message string is not null or not empty parse json message array to List of string messages
+        if (messagesJsonString != null && !messagesJsonString.trim().isEmpty()
+                && !("[]").equals(messagesJsonString.trim())) {
+
+            try {
+                // get root node
+                final JsonNode rootNode = objectMapper.readTree(messagesJsonString);
+                // iterate over root node and parse arrays messages
+                for (JsonNode jsonNode : rootNode) {
+                    // if array parse it is array of messages
+                    final String incomingMessageString = jsonNode.toString();
+                    if (jsonNode.isArray()) {
+                        final List messageList = objectMapper.readValue(incomingMessageString, List.class);
+                        for (Object message : messageList) {
+                            final String jsonMessageString = objectMapper.writeValueAsString(message);
+                            addUnescapedJsonToMessage(messages, jsonMessageString);
+                        }
+                    } else {
+                        // parse it as object
+                        addUnescapedJsonToMessage(messages, incomingMessageString);
+                    }
+                }
+
+            } catch (IOException e) {
+                final String errorMessage =
+                        format("Unable to convert subscriber Json String to Messages. Subscriber Response String: %s," +
+                                " Json Error: %s", messagesJsonString, e);
+                throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+            }
+
+        }
+        return messages;
+    }
+
+    /**
+     * Adds unescaped Json messages to given messages list
+     *
+     * @param messages message list in which unescaped messages will be added
+     * @param incomingMessageString incoming message string that may need to be escaped
+     */
+    private static void addUnescapedJsonToMessage(List<String> messages, String incomingMessageString) {
+        if (incomingMessageString.startsWith("\"") && incomingMessageString.endsWith("\"")) {
+            messages.add(StringEscapeUtils.unescapeJson(
+                    incomingMessageString.substring(1, incomingMessageString.length() - 1)));
+        } else {
+            messages.add(StringEscapeUtils.unescapeJson(incomingMessageString));
+        }
+    }
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
index 5c0df09..872c4fe 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/DMaaPMRComponent.java
@@ -1,29 +1,29 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service;
-
-/**
- * Marker interface for all DMaaP MR Components e.g. MR Publishers, MR Subscribers
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRComponent {
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service;
+
+/**
+ * Marker interface for all DMaaP MR Components e.g. MR Publishers, MR Subscribers
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRComponent {
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
index 48b0a70..04186c1 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisher.java
@@ -1,95 +1,95 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-
-import java.util.Date;
-import java.util.List;
-
-/**
- * <p>
- *     DMaaP MR Publisher can be used to publish messages to DMaaP MR Topics.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRPublisher extends AutoCloseable {
-
-
-    /**
-     * <p>
-     *     Adds collection of messages to DMaaP MR Topic Publishing Queue.
-     * <p>
-     *     Note: Invoking this method may or may not cause publishing immediately
-     *     as publishing in done is batch mode by default. Parameter maxBatchSize
-     *     in {@link DMaaPMRPublisherConfig} is used to determine max batch queue size.
-     *     If the maxBatchSize is reached all message will be published automatically
-     *     during subsequent call.
-     * </p>
-     *
-     * @param messages messages to publish to DMaaP MR Publisher
-     * @return response which may contain Http Response code 202 (Accepted) as publishing
-     * will proceed when max batch size is reached. Throws {@link DCAEAnalyticsRuntimeException}
-     * if publishing fails
-     */
-    DMaaPMRPublisherResponse publish(List<String> messages);
-
-
-    /**
-     * <p>
-     *     Forces publishing of messages to DMaaP MR Topic and returns {@link DMaaPMRPublisherResponse}
-     *     which can be inspected for HTTP status code of publishing call to DMaaP MR Topic.
-     * </p>
-     *
-     * @param messages messages to publish to DMaaP MR Publisher
-     * @return DMaaP Message Router Publisher Response. Throws {@link DCAEAnalyticsRuntimeException}
-     * if force publishing fails
-     *
-     */
-    DMaaPMRPublisherResponse forcePublish(List<String> messages);
-
-
-    /**
-     * <p>
-     *     Forces publishing of messages in Publisher queue to DMaaP MR Topic and returns
-     *     {@link DMaaPMRPublisherResponse}.If there are no messages were in the queue to
-     *     be flushed response code 304 (Not Modified) will be returned
-     * </p>
-     *
-     * @return DMaaP Message Router Publisher Response
-     */
-    DMaaPMRPublisherResponse flush();
-
-
-    /**
-     * <p>
-     *     Returns the creation time when Publisher instance was created.
-     * <p>
-     *
-     * @return creation time of Subscriber instance
-     */
-    Date getPublisherCreationTime();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+
+import java.util.Date;
+import java.util.List;
+
+/**
+ * <p>
+ *     DMaaP MR Publisher can be used to publish messages to DMaaP MR Topics.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRPublisher extends AutoCloseable {
+
+
+    /**
+     * <p>
+     *     Adds collection of messages to DMaaP MR Topic Publishing Queue.
+     * <p>
+     *     Note: Invoking this method may or may not cause publishing immediately
+     *     as publishing in done is batch mode by default. Parameter maxBatchSize
+     *     in {@link DMaaPMRPublisherConfig} is used to determine max batch queue size.
+     *     If the maxBatchSize is reached all message will be published automatically
+     *     during subsequent call.
+     * </p>
+     *
+     * @param messages messages to publish to DMaaP MR Publisher
+     * @return response which may contain Http Response code 202 (Accepted) as publishing
+     * will proceed when max batch size is reached. Throws {@link DCAEAnalyticsRuntimeException}
+     * if publishing fails
+     */
+    DMaaPMRPublisherResponse publish(List<String> messages);
+
+
+    /**
+     * <p>
+     *     Forces publishing of messages to DMaaP MR Topic and returns {@link DMaaPMRPublisherResponse}
+     *     which can be inspected for HTTP status code of publishing call to DMaaP MR Topic.
+     * </p>
+     *
+     * @param messages messages to publish to DMaaP MR Publisher
+     * @return DMaaP Message Router Publisher Response. Throws {@link DCAEAnalyticsRuntimeException}
+     * if force publishing fails
+     *
+     */
+    DMaaPMRPublisherResponse forcePublish(List<String> messages);
+
+
+    /**
+     * <p>
+     *     Forces publishing of messages in Publisher queue to DMaaP MR Topic and returns
+     *     {@link DMaaPMRPublisherResponse}.If there are no messages were in the queue to
+     *     be flushed response code 304 (Not Modified) will be returned
+     * </p>
+     *
+     * @return DMaaP Message Router Publisher Response
+     */
+    DMaaPMRPublisherResponse flush();
+
+
+    /**
+     * <p>
+     *     Returns the creation time when Publisher instance was created.
+     * <p>
+     *
+     * @return creation time of Subscriber instance
+     */
+    Date getPublisherCreationTime();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
index 5d6dfa1..3b55b57 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherFactory.java
@@ -1,49 +1,49 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-
-/**
- * <p>
- *     Factory to initialize instance of {@link DMaaPMRPublisher} for Guice DI injection purposes.
- * <p>
- *     <strong>
- *          NOTE: Client should not use this Factory to initialize {@link DMaaPMRPublisher} unless they
- *                are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
- *                guice injected Publisher instances
- *      </strong>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public interface DMaaPMRPublisherFactory {
-
-    /**
-     * Guice Factory to create DMaaP MR Publisher
-     *
-     * @param publisherConfig publisher config
-     *
-     * @return DMaaP MR Publisher instance
-     */
-    DMaaPMRPublisher create(DMaaPMRPublisherConfig publisherConfig);
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+
+/**
+ * <p>
+ *     Factory to initialize instance of {@link DMaaPMRPublisher} for Guice DI injection purposes.
+ * <p>
+ *     <strong>
+ *          NOTE: Client should not use this Factory to initialize {@link DMaaPMRPublisher} unless they
+ *                are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
+ *                guice injected Publisher instances
+ *      </strong>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public interface DMaaPMRPublisherFactory {
+
+    /**
+     * Guice Factory to create DMaaP MR Publisher
+     *
+     * @param publisherConfig publisher config
+     *
+     * @return DMaaP MR Publisher instance
+     */
+    DMaaPMRPublisher create(DMaaPMRPublisherConfig publisherConfig);
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
index 32e09fa..16ea5fa 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherImpl.java
@@ -1,210 +1,210 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.common.base.Optional;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpHeaders;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ContentType;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.Date;
-import java.util.List;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
-import static java.lang.String.format;
-
-/**
- * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
-
-    private final DMaaPMRPublisherConfig publisherConfig;
-    private final CloseableHttpClient closeableHttpClient;
-    private final DMaaPMRPublisherQueue publisherQueue;
-    private final Date publisherCreationTime;
-    private URI publisherUri;
-
-    @Inject
-    public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
-                                DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
-                                CloseableHttpClient closeableHttpClient) {
-
-        this.publisherConfig = publisherConfig;
-        final int maxBatchSize = publisherConfig.getMaxBatchSize() > 0 ? publisherConfig.getMaxBatchSize() : 1;
-        this.publisherQueue = dMaaPMRPublisherQueueFactory.create(
-                maxBatchSize, publisherConfig.getMaxRecoveryQueueSize());
-        this.closeableHttpClient = closeableHttpClient;
-        this.publisherUri = createPublisherURI(publisherConfig);
-        this.publisherCreationTime = new Date();
-    }
-
-
-    @Override
-    public DMaaPMRPublisherResponse publish(List<String> messages)  {
-
-        final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
-
-        // if messages size is less than batch queue size - just queue them for batch publishing
-        if (batchQueueRemainingSize > messages.size()) {
-            LOG.debug("Adding messages to batch Queue. No flushing required. Messages Size:{}. Batch Queue Size:{}",
-                    messages.size(), batchQueueRemainingSize);
-            final int batchQueueSize = publisherQueue.addBatchMessages(messages);
-            return createPublisherAcceptedResponse(batchQueueSize);
-
-        } else {
-
-            // grab all already queued messages, append current messages and force publish them to DMaaP MR topic
-            final List<String> queueMessages = publisherQueue.getMessageForPublishing();
-            LOG.debug("Batch Queue capacity exceeds messages size. Flushing of all pending messages to DMaaP MR " +
-                    "Publisher Topic.");
-            return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
-        }
-
-    }
-
-    @Override
-    public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
-
-        LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
-
-        final String contentType = publisherConfig.getContentType();
-        final String userName = publisherConfig.getUserName();
-        final String userPassword = publisherConfig.getUserPassword();
-        final HttpPost postRequest = new HttpPost(publisherUri);
-
-        // add Authorization Header if username and password are present
-        final Optional<String> authHeader = getAuthHeader(userName, userPassword);
-        if (authHeader.isPresent()) {
-            postRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
-        } else {
-            LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
-        }
-
-        // Create post string entity
-        final String messagesJson = convertToJsonString(messages);
-        final StringEntity requestEntity =
-                new StringEntity(messagesJson, ContentType.create(contentType, "UTF-8"));
-        postRequest.setEntity(requestEntity);
-
-        try {
-            final Pair<Integer, String> responsePair = closeableHttpClient.execute(postRequest, responseHandler());
-            final Integer responseCode = responsePair.getLeft();
-            final String responseBody = responsePair.getRight();
-            // if messages were published successfully, return successful response
-            if (isSuccessfulResponseCode(responseCode)) {
-                LOG.debug("DMaaP MR Messages published successfully. DMaaP Response Code: {}. DMaaP Response " +
-                                "Body: {}, Number of Messages published: {}",
-                        responseCode, responseBody, messages.size());
-
-            } else {
-                LOG.warn("Unable to publish messages to DMaaP MR Topic. DMaaP Response Code: {}, DMaaP Response " +
-                        "Body: {}. Messages will be queued in recovery queue", responseCode, responseBody);
-                addMessagesToRecoveryQueue(publisherQueue, messages);
-            }
-
-            return createPublisherResponse(responseCode, responseBody,
-                    getPendingMessages(publisherQueue, publisherConfig));
-
-        } catch (IOException e) {
-            // If IO Error then we need to also put messages in recovery queue
-            addMessagesToRecoveryQueue(publisherQueue, messages);
-            final String errorMessage = format("IO Exception while publishing messages to DMaaP Topic. " +
-                    "Messages will be queued in recovery queue. Messages Size: %d", messages.size());
-
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-
-    }
-
-
-    @Override
-    public DMaaPMRPublisherResponse flush() {
-        final List<String> queueMessages = publisherQueue.getMessageForPublishing();
-        // If there are no message return 204 (No Content) response code
-        if (queueMessages.isEmpty()) {
-            LOG.debug("No messages to publish to batch queue. Returning 204 status code");
-            return createPublisherNoContentResponse();
-        } else {
-            // force publish messages in queue
-            return forcePublish(queueMessages);
-        }
-    }
-
-    @Override
-    public Date getPublisherCreationTime() {
-        return new Date(publisherCreationTime.getTime());
-    }
-
-    @Override
-    public void close() throws Exception {
-
-        // flush current message in the queue
-        int retrialNumber = 0;
-        int flushResponseCode;
-
-        // automatic retries if messages cannot be flushed
-        do {
-            retrialNumber++;
-            DMaaPMRPublisherResponse flushResponse = flush();
-            flushResponseCode = flushResponse.getResponseCode();
-
-            if (!isSuccessfulResponseCode(flushResponseCode)) {
-                LOG.warn("Unable to flush batch messages to publisher due to DMaaP MR invalid Response: {}. " +
-                                "Retrial No: {} of Max {} Retries", flushResponseCode, retrialNumber,
-                        AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE);
-
-                Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
-            }
-        } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
-                !isSuccessfulResponseCode(flushResponseCode));
-
-        if (!isSuccessfulResponseCode(flushResponseCode)) {
-            LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
-        } else {
-            LOG.info("Successfully published all batched messages to publisher.");
-        }
-
-        // close http client
-        closeableHttpClient.close();
-
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRPublisherResponse;
+import org.onap.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Date;
+import java.util.List;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
+import static java.lang.String.format;
+
+/**
+ * Concrete Implementation of {@link DMaaPMRPublisher} which uses {@link HttpClient}
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRPublisherImpl extends BaseDMaaPMRComponent implements DMaaPMRPublisher {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherImpl.class);
+
+    private final DMaaPMRPublisherConfig publisherConfig;
+    private final CloseableHttpClient closeableHttpClient;
+    private final DMaaPMRPublisherQueue publisherQueue;
+    private final Date publisherCreationTime;
+    private URI publisherUri;
+
+    @Inject
+    public DMaaPMRPublisherImpl(@Assisted DMaaPMRPublisherConfig publisherConfig,
+                                DMaaPMRPublisherQueueFactory dMaaPMRPublisherQueueFactory,
+                                CloseableHttpClient closeableHttpClient) {
+
+        this.publisherConfig = publisherConfig;
+        final int maxBatchSize = publisherConfig.getMaxBatchSize() > 0 ? publisherConfig.getMaxBatchSize() : 1;
+        this.publisherQueue = dMaaPMRPublisherQueueFactory.create(
+                maxBatchSize, publisherConfig.getMaxRecoveryQueueSize());
+        this.closeableHttpClient = closeableHttpClient;
+        this.publisherUri = createPublisherURI(publisherConfig);
+        this.publisherCreationTime = new Date();
+    }
+
+
+    @Override
+    public DMaaPMRPublisherResponse publish(List<String> messages)  {
+
+        final int batchQueueRemainingSize = publisherQueue.getBatchQueueRemainingSize();
+
+        // if messages size is less than batch queue size - just queue them for batch publishing
+        if (batchQueueRemainingSize > messages.size()) {
+            LOG.debug("Adding messages to batch Queue. No flushing required. Messages Size:{}. Batch Queue Size:{}",
+                    messages.size(), batchQueueRemainingSize);
+            final int batchQueueSize = publisherQueue.addBatchMessages(messages);
+            return createPublisherAcceptedResponse(batchQueueSize);
+
+        } else {
+
+            // grab all already queued messages, append current messages and force publish them to DMaaP MR topic
+            final List<String> queueMessages = publisherQueue.getMessageForPublishing();
+            LOG.debug("Batch Queue capacity exceeds messages size. Flushing of all pending messages to DMaaP MR " +
+                    "Publisher Topic.");
+            return forcePublish(Lists.newLinkedList(Iterables.concat(queueMessages, messages)));
+        }
+
+    }
+
+    @Override
+    public DMaaPMRPublisherResponse forcePublish(List<String> messages) {
+
+        LOG.debug("Force publishing messages to DMaaP MR Topic. Messages Size: {}", messages.size());
+
+        final String contentType = publisherConfig.getContentType();
+        final String userName = publisherConfig.getUserName();
+        final String userPassword = publisherConfig.getUserPassword();
+        final HttpPost postRequest = new HttpPost(publisherUri);
+
+        // add Authorization Header if username and password are present
+        final Optional<String> authHeader = getAuthHeader(userName, userPassword);
+        if (authHeader.isPresent()) {
+            postRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
+        } else {
+            LOG.debug("DMaaP MR Publisher Authentication is disabled as username or password is not present.");
+        }
+
+        // Create post string entity
+        final String messagesJson = convertToJsonString(messages);
+        final StringEntity requestEntity =
+                new StringEntity(messagesJson, ContentType.create(contentType, "UTF-8"));
+        postRequest.setEntity(requestEntity);
+
+        try {
+            final Pair<Integer, String> responsePair = closeableHttpClient.execute(postRequest, responseHandler());
+            final Integer responseCode = responsePair.getLeft();
+            final String responseBody = responsePair.getRight();
+            // if messages were published successfully, return successful response
+            if (isSuccessfulResponseCode(responseCode)) {
+                LOG.debug("DMaaP MR Messages published successfully. DMaaP Response Code: {}. DMaaP Response " +
+                                "Body: {}, Number of Messages published: {}",
+                        responseCode, responseBody, messages.size());
+
+            } else {
+                LOG.warn("Unable to publish messages to DMaaP MR Topic. DMaaP Response Code: {}, DMaaP Response " +
+                        "Body: {}. Messages will be queued in recovery queue", responseCode, responseBody);
+                addMessagesToRecoveryQueue(publisherQueue, messages);
+            }
+
+            return createPublisherResponse(responseCode, responseBody,
+                    getPendingMessages(publisherQueue, publisherConfig));
+
+        } catch (IOException e) {
+            // If IO Error then we need to also put messages in recovery queue
+            addMessagesToRecoveryQueue(publisherQueue, messages);
+            final String errorMessage = format("IO Exception while publishing messages to DMaaP Topic. " +
+                    "Messages will be queued in recovery queue. Messages Size: %d", messages.size());
+
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+
+    }
+
+
+    @Override
+    public DMaaPMRPublisherResponse flush() {
+        final List<String> queueMessages = publisherQueue.getMessageForPublishing();
+        // If there are no message return 204 (No Content) response code
+        if (queueMessages.isEmpty()) {
+            LOG.debug("No messages to publish to batch queue. Returning 204 status code");
+            return createPublisherNoContentResponse();
+        } else {
+            // force publish messages in queue
+            return forcePublish(queueMessages);
+        }
+    }
+
+    @Override
+    public Date getPublisherCreationTime() {
+        return new Date(publisherCreationTime.getTime());
+    }
+
+    @Override
+    public void close() throws Exception {
+
+        // flush current message in the queue
+        int retrialNumber = 0;
+        int flushResponseCode;
+
+        // automatic retries if messages cannot be flushed
+        do {
+            retrialNumber++;
+            DMaaPMRPublisherResponse flushResponse = flush();
+            flushResponseCode = flushResponse.getResponseCode();
+
+            if (!isSuccessfulResponseCode(flushResponseCode)) {
+                LOG.warn("Unable to flush batch messages to publisher due to DMaaP MR invalid Response: {}. " +
+                                "Retrial No: {} of Max {} Retries", flushResponseCode, retrialNumber,
+                        AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE);
+
+                Thread.sleep(AnalyticsConstants.PUBLISHER_DELAY_MS_ON_RETRIES_ON_CLOSE);
+            }
+        } while (retrialNumber <= AnalyticsConstants.PUBLISHER_MAX_FLUSH_RETRIES_ON_CLOSE &&
+                !isSuccessfulResponseCode(flushResponseCode));
+
+        if (!isSuccessfulResponseCode(flushResponseCode)) {
+            LOG.error("Unable to flush batch messages to publisher. Messages loss cannot be prevented");
+        } else {
+            LOG.info("Successfully published all batched messages to publisher.");
+        }
+
+        // close http client
+        closeableHttpClient.close();
+
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
index 3877f0b..e625fba 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueue.java
@@ -1,87 +1,87 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import java.util.List;
-
-/**
- * <p>
- *     DMaaP MR Publisher Queue handles back pressure in case DMaaP MR Publisher topic
- *     is offline for some reason. It does so by having a recovery queue which keeps
- *     messages in order in case there is temporary interruption in DMaaP Publisher
- * </p>
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRPublisherQueue {
-
-    /**
-     * <p>
-     *     Add batchMessages to Batch Queue
-     * </p>
-     *
-     * @param batchMessages messages that needs to be added to batch queue
-     * @return current size of batch queue. Throws {@link IllegalStateException}
-     * if batch queue does not have enough space
-     */
-    int addBatchMessages(List<String> batchMessages);
-
-
-    /**
-     * <p>
-     *     Add recoverable messages to Recoverable Queue
-     * </p>
-     *
-     * @param recoverableMessages messages that needs to be added to recoverable queue
-     * @return current size of the recoverable queue. Throws {@link IllegalStateException}
-     * if recoverable queue does not have enough space
-     */
-    int addRecoverableMessages(List<String> recoverableMessages);
-
-    /**
-     * <p>
-     *     Get messages that need to be published to DMaaP topic. Messages in recoverable
-     *     queue are appended if present.
-     * </p>
-     *
-     * @return List of messages from both batch and recovery queue
-     */
-    List<String> getMessageForPublishing();
-
-    /**
-     * <p>
-     *     Remaining capacity of Batch Queue
-     * </p>
-     *
-     * @return Remaining Batch Queue Size
-     */
-    int getBatchQueueRemainingSize();
-
-    /**
-     * <p>
-     *     Remaining capacity of Recovery Queue
-     * </p>
-     *
-     * @return Remaining Recovery Queue Size
-     */
-    int getRecoveryQueueRemainingSize();
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import java.util.List;
+
+/**
+ * <p>
+ *     DMaaP MR Publisher Queue handles back pressure in case DMaaP MR Publisher topic
+ *     is offline for some reason. It does so by having a recovery queue which keeps
+ *     messages in order in case there is temporary interruption in DMaaP Publisher
+ * </p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRPublisherQueue {
+
+    /**
+     * <p>
+     *     Add batchMessages to Batch Queue
+     * </p>
+     *
+     * @param batchMessages messages that needs to be added to batch queue
+     * @return current size of batch queue. Throws {@link IllegalStateException}
+     * if batch queue does not have enough space
+     */
+    int addBatchMessages(List<String> batchMessages);
+
+
+    /**
+     * <p>
+     *     Add recoverable messages to Recoverable Queue
+     * </p>
+     *
+     * @param recoverableMessages messages that needs to be added to recoverable queue
+     * @return current size of the recoverable queue. Throws {@link IllegalStateException}
+     * if recoverable queue does not have enough space
+     */
+    int addRecoverableMessages(List<String> recoverableMessages);
+
+    /**
+     * <p>
+     *     Get messages that need to be published to DMaaP topic. Messages in recoverable
+     *     queue are appended if present.
+     * </p>
+     *
+     * @return List of messages from both batch and recovery queue
+     */
+    List<String> getMessageForPublishing();
+
+    /**
+     * <p>
+     *     Remaining capacity of Batch Queue
+     * </p>
+     *
+     * @return Remaining Batch Queue Size
+     */
+    int getBatchQueueRemainingSize();
+
+    /**
+     * <p>
+     *     Remaining capacity of Recovery Queue
+     * </p>
+     *
+     * @return Remaining Recovery Queue Size
+     */
+    int getRecoveryQueueRemainingSize();
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
index 8d44f93..aeacbb0 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueFactory.java
@@ -1,45 +1,45 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.inject.assistedinject.Assisted;
-
-/**
- * <p>
- *     Factory to initialize instance of {@link DMaaPMRPublisherQueue} for Guice DI injection purposes.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public interface DMaaPMRPublisherQueueFactory {
-
-    /**
-     * Guice Factory to create DMaaP MR Publisher Queue
-     *
-     * @param batchQueueSize batch queue size
-     * @param recoveryQueueSize recovery queue size
-     *
-     * @return instance of DMaaP MR Publisher Queue
-     */
-    DMaaPMRPublisherQueue create(@Assisted("batchQueueSize") int batchQueueSize,
-                                 @Assisted("recoveryQueueSize") int recoveryQueueSize);
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.inject.assistedinject.Assisted;
+
+/**
+ * <p>
+ *     Factory to initialize instance of {@link DMaaPMRPublisherQueue} for Guice DI injection purposes.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public interface DMaaPMRPublisherQueueFactory {
+
+    /**
+     * Guice Factory to create DMaaP MR Publisher Queue
+     *
+     * @param batchQueueSize batch queue size
+     * @param recoveryQueueSize recovery queue size
+     *
+     * @return instance of DMaaP MR Publisher Queue
+     */
+    DMaaPMRPublisherQueue create(@Assisted("batchQueueSize") int batchQueueSize,
+                                 @Assisted("recoveryQueueSize") int recoveryQueueSize);
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
index e42b6b0..343c31a 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/publisher/DMaaPMRPublisherQueueImpl.java
@@ -1,126 +1,126 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.publisher;
-
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.LinkedBlockingDeque;
-
-import static com.google.common.collect.Iterables.concat;
-import static com.google.common.collect.Lists.newLinkedList;
-import static java.util.Collections.unmodifiableList;
-
-/**
- * <p>
- *     An implementation of {@link DMaaPMRPublisherQueue} which uses {@link java.util.concurrent.BlockingDeque}
- *     for batch and recovery queues
- * </p>
- *
- *
- * @author Rajiv Singla . Creation Date: 11/1/2016.
- */
-public class DMaaPMRPublisherQueueImpl implements DMaaPMRPublisherQueue {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherQueueImpl.class);
-
-    private final LinkedBlockingDeque<String> batchQueue;
-    private final LinkedBlockingDeque<String> recoveryQueue;
-
-    @Inject
-    public DMaaPMRPublisherQueueImpl(@Assisted("batchQueueSize") int batchQueueSize,
-                                     @Assisted("recoveryQueueSize") int recoveryQueueSize) {
-        batchQueue = new LinkedBlockingDeque<>(batchQueueSize);
-        recoveryQueue = new LinkedBlockingDeque<>(recoveryQueueSize);
-        LOG.debug("Creating Instance of DMaaP Publisher Queue. BatchQueueSize: {}, RecoveryQueueSize: {}",
-                batchQueueSize, recoveryQueueSize);
-    }
-
-    @Override
-    public synchronized int addBatchMessages(List<String> batchMessages) {
-
-        // checks if batchMessages size does not exceed batch queue capacity
-        if (batchMessages.size() > batchQueue.remainingCapacity()) {
-            throw new IllegalStateException("Not enough capacity to add batchMessages  in batch queue");
-        }
-
-        // Add batchMessages to batch queue
-        for (String message : batchMessages) {
-            batchQueue.add(message);
-        }
-
-        // returns current elements size in batch queue
-        return batchQueue.size();
-    }
-
-    @Override
-    public synchronized int addRecoverableMessages(List<String> recoverableMessages) {
-
-        // checks if messages size does not exceed recovery queue size
-        if (recoverableMessages.size() > recoveryQueue.remainingCapacity()) {
-            throw new IllegalStateException("Not enough capacity to add messages in recovery queue");
-        }
-
-        // add messages to recovery queue
-        for (String recoverableMessage : recoverableMessages) {
-            recoveryQueue.add(recoverableMessage);
-        }
-
-        // returns current size of recovery queue
-        return recoveryQueue.size();
-    }
-
-    @Override
-    public synchronized List<String> getMessageForPublishing() {
-
-        final List<String> recoveryMessageList = new LinkedList<>();
-        final List<String> batchMessagesList = new LinkedList<>();
-
-        // get messages from recovery queue if present
-        if (!recoveryQueue.isEmpty()) {
-            final int recoveryQueueSize = recoveryQueue.drainTo(recoveryMessageList);
-            LOG.debug("Drained Recovery Queue elements for flushing: {}", recoveryQueueSize);
-        }
-
-        // get messages from batch queue if present
-        if (!batchQueue.isEmpty()) {
-            final int batchQueueSize = batchQueue.drainTo(batchMessagesList);
-            LOG.debug("Drained Batch Queue elements for flushing: {}", batchQueueSize);
-        }
-
-        // concat recovery and batch queue elements
-        return unmodifiableList(newLinkedList(concat(recoveryMessageList, batchMessagesList)));
-    }
-
-    @Override
-    public synchronized int getBatchQueueRemainingSize() {
-        return batchQueue.remainingCapacity();
-    }
-
-    @Override
-    public synchronized int getRecoveryQueueRemainingSize() {
-        return recoveryQueue.remainingCapacity();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.publisher;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import static com.google.common.collect.Iterables.concat;
+import static com.google.common.collect.Lists.newLinkedList;
+import static java.util.Collections.unmodifiableList;
+
+/**
+ * <p>
+ *     An implementation of {@link DMaaPMRPublisherQueue} which uses {@link java.util.concurrent.BlockingDeque}
+ *     for batch and recovery queues
+ * </p>
+ *
+ *
+ * @author Rajiv Singla . Creation Date: 11/1/2016.
+ */
+public class DMaaPMRPublisherQueueImpl implements DMaaPMRPublisherQueue {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRPublisherQueueImpl.class);
+
+    private final LinkedBlockingDeque<String> batchQueue;
+    private final LinkedBlockingDeque<String> recoveryQueue;
+
+    @Inject
+    public DMaaPMRPublisherQueueImpl(@Assisted("batchQueueSize") int batchQueueSize,
+                                     @Assisted("recoveryQueueSize") int recoveryQueueSize) {
+        batchQueue = new LinkedBlockingDeque<>(batchQueueSize);
+        recoveryQueue = new LinkedBlockingDeque<>(recoveryQueueSize);
+        LOG.debug("Creating Instance of DMaaP Publisher Queue. BatchQueueSize: {}, RecoveryQueueSize: {}",
+                batchQueueSize, recoveryQueueSize);
+    }
+
+    @Override
+    public synchronized int addBatchMessages(List<String> batchMessages) {
+
+        // checks if batchMessages size does not exceed batch queue capacity
+        if (batchMessages.size() > batchQueue.remainingCapacity()) {
+            throw new IllegalStateException("Not enough capacity to add batchMessages  in batch queue");
+        }
+
+        // Add batchMessages to batch queue
+        for (String message : batchMessages) {
+            batchQueue.add(message);
+        }
+
+        // returns current elements size in batch queue
+        return batchQueue.size();
+    }
+
+    @Override
+    public synchronized int addRecoverableMessages(List<String> recoverableMessages) {
+
+        // checks if messages size does not exceed recovery queue size
+        if (recoverableMessages.size() > recoveryQueue.remainingCapacity()) {
+            throw new IllegalStateException("Not enough capacity to add messages in recovery queue");
+        }
+
+        // add messages to recovery queue
+        for (String recoverableMessage : recoverableMessages) {
+            recoveryQueue.add(recoverableMessage);
+        }
+
+        // returns current size of recovery queue
+        return recoveryQueue.size();
+    }
+
+    @Override
+    public synchronized List<String> getMessageForPublishing() {
+
+        final List<String> recoveryMessageList = new LinkedList<>();
+        final List<String> batchMessagesList = new LinkedList<>();
+
+        // get messages from recovery queue if present
+        if (!recoveryQueue.isEmpty()) {
+            final int recoveryQueueSize = recoveryQueue.drainTo(recoveryMessageList);
+            LOG.debug("Drained Recovery Queue elements for flushing: {}", recoveryQueueSize);
+        }
+
+        // get messages from batch queue if present
+        if (!batchQueue.isEmpty()) {
+            final int batchQueueSize = batchQueue.drainTo(batchMessagesList);
+            LOG.debug("Drained Batch Queue elements for flushing: {}", batchQueueSize);
+        }
+
+        // concat recovery and batch queue elements
+        return unmodifiableList(newLinkedList(concat(recoveryMessageList, batchMessagesList)));
+    }
+
+    @Override
+    public synchronized int getBatchQueueRemainingSize() {
+        return batchQueue.remainingCapacity();
+    }
+
+    @Override
+    public synchronized int getRecoveryQueueRemainingSize() {
+        return recoveryQueue.remainingCapacity();
+    }
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
index f193b2c..6a0aa63 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriber.java
@@ -1,57 +1,57 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-
-import java.util.Date;
-
-/**
- * <p>
- *     DMaaP MR Subscriber can be used to subscribe messages from DMaaP MR Topics.
- * <p>
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public interface DMaaPMRSubscriber extends AutoCloseable {
-
-    /**
-     * Fetches Messages from DMaaP MR Topic. {@link DMaaPMRPublisherConfig} settings parameters
-     * for messageLimit and message timeout are used
-     *
-     * @return DMaaP Message Router Subscriber Response
-     */
-    DMaaPMRSubscriberResponse fetchMessages();
-
-
-    /**
-     * Returns the Subscriber instance creation time
-     * <p>
-     * NOTE: Due to DMaaP API Design - Subscribers can only fetch messages which
-     * are published to the topic after the creation of the Subscriber.
-     *
-     * @return creation time of Subscriber instance
-     */
-    Date getSubscriberCreationTime();
-
-
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.subscriber;
+
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+
+import java.util.Date;
+
+/**
+ * <p>
+ *     DMaaP MR Subscriber can be used to subscribe messages from DMaaP MR Topics.
+ * <p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public interface DMaaPMRSubscriber extends AutoCloseable {
+
+    /**
+     * Fetches Messages from DMaaP MR Topic. {@link DMaaPMRPublisherConfig} settings parameters
+     * for messageLimit and message timeout are used
+     *
+     * @return DMaaP Message Router Subscriber Response
+     */
+    DMaaPMRSubscriberResponse fetchMessages();
+
+
+    /**
+     * Returns the Subscriber instance creation time
+     * <p>
+     * NOTE: Due to DMaaP API Design - Subscribers can only fetch messages which
+     * are published to the topic after the creation of the Subscriber.
+     *
+     * @return creation time of Subscriber instance
+     */
+    Date getSubscriberCreationTime();
+
+
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
index b7f5ada..e06e2bd 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberFactory.java
@@ -1,47 +1,47 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-
-/**
- * Factory to initialize instance of {@link DMaaPMRSubscriber} for Guice DI injection purposes.
- * <p>
- *     <strong>
- *          NOTE: Client should not use this Factory to initialize {@link DMaaPMRSubscriber} unless they
- *              are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
- *              guice injected Subscriber instances
- *      </strong>
- * <p>
- * @author Rajiv Singla . Creation Date: 10/20/2016.
- */
-public interface DMaaPMRSubscriberFactory {
-
-    /**
-     * Guice Factory to create DMaaP MR Subscriber Instance
-     *
-     * @param subscriberConfig subscriber config
-     *
-     * @return DMaaP MR Subscriber instance
-     */
-    DMaaPMRSubscriber create(DMaaPMRSubscriberConfig subscriberConfig);
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.subscriber;
+
+import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+
+/**
+ * Factory to initialize instance of {@link DMaaPMRSubscriber} for Guice DI injection purposes.
+ * <p>
+ *     <strong>
+ *          NOTE: Client should not use this Factory to initialize {@link DMaaPMRSubscriber} unless they
+ *              are wiring dependencies using Guice. Client must use {@link DMaaPMRFactory} to initialize
+ *              guice injected Subscriber instances
+ *      </strong>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 10/20/2016.
+ */
+public interface DMaaPMRSubscriberFactory {
+
+    /**
+     * Guice Factory to create DMaaP MR Subscriber Instance
+     *
+     * @param subscriberConfig subscriber config
+     *
+     * @return DMaaP MR Subscriber instance
+     */
+    DMaaPMRSubscriber create(DMaaPMRSubscriberConfig subscriberConfig);
+}
diff --git a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
index 678ff76..9d75ffc 100644
--- a/dcae-analytics-dmaap/src/main/java/org/openecomp/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
+++ b/dcae-analytics-dmaap/src/main/java/org/onap/dcae/apod/analytics/dmaap/service/subscriber/DMaaPMRSubscriberImpl.java
@@ -1,129 +1,129 @@
-/*
- * ===============================LICENSE_START======================================
- *  dcae-analytics
- * ================================================================================
- *    Copyright © 2017 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.openecomp.dcae.apod.analytics.dmaap.service.subscriber;
-
-import com.google.common.base.Optional;
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.http.HttpHeaders;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
-import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
-import org.openecomp.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.Date;
-import java.util.LinkedList;
-import java.util.List;
-
-import static org.openecomp.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
-import static java.lang.String.format;
-
-/**
- * Concrete Implementation of {@link DMaaPMRSubscriber} which uses {@link HttpClient}
- *
- * @author Rajiv Singla . Creation Date: 10/13/2016.
- */
-public class DMaaPMRSubscriberImpl extends BaseDMaaPMRComponent implements DMaaPMRSubscriber {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSubscriberImpl.class);
-
-    private final DMaaPMRSubscriberConfig subscriberConfig;
-    private final CloseableHttpClient closeableHttpClient;
-    private final URI subscriberUri;
-    private final Date subscriberCreationTime;
-
-    @Inject
-    public DMaaPMRSubscriberImpl(@Assisted DMaaPMRSubscriberConfig subscriberConfig,
-                                 CloseableHttpClient closeableHttpClient) {
-        this.subscriberConfig = subscriberConfig;
-        this.closeableHttpClient = closeableHttpClient;
-        this.subscriberUri = createSubscriberURI(subscriberConfig);
-        this.subscriberCreationTime = new Date();
-    }
-
-    @Override
-    public DMaaPMRSubscriberResponse fetchMessages() {
-
-        final String userName = subscriberConfig.getUserName();
-        final String userPassword = subscriberConfig.getUserPassword();
-
-        final HttpGet getRequest = new HttpGet(subscriberUri);
-
-        // add Authorization Header if username and password are present
-        final Optional<String> authHeader = getAuthHeader(userName, userPassword);
-        if (authHeader.isPresent()) {
-            getRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
-        } else {
-            LOG.debug("DMaaP MR Subscriber Authentication is disabled as username or password is not present.");
-        }
-
-        try {
-
-            final Pair<Integer, String> responsePair = closeableHttpClient.execute(getRequest, responseHandler());
-            final Integer responseCode = responsePair.getLeft();
-            final String responseBody = responsePair.getRight();
-
-            List<String> fetchedMessages = new LinkedList<>();
-            String responseMessage = responseBody;
-
-            // if messages were published successfully, return successful response
-            if (isSuccessfulResponseCode(responseCode)) {
-                if (responseBody != null) {
-                    fetchedMessages = convertJsonToStringMessages(responseBody);
-                    responseMessage = "Messages Fetched Successfully";
-                } else {
-                    responseMessage = "DMaaP Response Body had no messages";
-                }
-            } else {
-                LOG.error("Unable to fetch messages to DMaaP MR Topic. DMaaP MR unsuccessful Response Code: {}, " +
-                        "DMaaP Response Body: {}", responseCode, responseBody);
-            }
-
-            return createSubscriberResponse(responseCode, responseMessage, fetchedMessages);
-
-        } catch (IOException e) {
-
-            final String errorMessage =
-                    format("IO Exception while fetching messages from DMaaP Topic. Exception %s", e);
-            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-        }
-
-
-    }
-
-    @Override
-    public Date getSubscriberCreationTime() {
-        return new Date(subscriberCreationTime.getTime());
-    }
-
-    @Override
-    public void close() throws Exception {
-        closeableHttpClient.close();
-    }
-}
+/*
+ * ===============================LICENSE_START======================================
+ *  dcae-analytics
+ * ================================================================================
+ *    Copyright © 2017 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.dcae.apod.analytics.dmaap.service.subscriber;
+
+import com.google.common.base.Optional;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.http.HttpHeaders;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRSubscriberConfig;
+import org.onap.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.onap.dcae.apod.analytics.dmaap.service.BaseDMaaPMRComponent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+
+import static org.onap.dcae.apod.analytics.common.utils.HTTPUtils.isSuccessfulResponseCode;
+import static java.lang.String.format;
+
+/**
+ * Concrete Implementation of {@link DMaaPMRSubscriber} which uses {@link HttpClient}
+ *
+ * @author Rajiv Singla . Creation Date: 10/13/2016.
+ */
+public class DMaaPMRSubscriberImpl extends BaseDMaaPMRComponent implements DMaaPMRSubscriber {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSubscriberImpl.class);
+
+    private final DMaaPMRSubscriberConfig subscriberConfig;
+    private final CloseableHttpClient closeableHttpClient;
+    private final URI subscriberUri;
+    private final Date subscriberCreationTime;
+
+    @Inject
+    public DMaaPMRSubscriberImpl(@Assisted DMaaPMRSubscriberConfig subscriberConfig,
+                                 CloseableHttpClient closeableHttpClient) {
+        this.subscriberConfig = subscriberConfig;
+        this.closeableHttpClient = closeableHttpClient;
+        this.subscriberUri = createSubscriberURI(subscriberConfig);
+        this.subscriberCreationTime = new Date();
+    }
+
+    @Override
+    public DMaaPMRSubscriberResponse fetchMessages() {
+
+        final String userName = subscriberConfig.getUserName();
+        final String userPassword = subscriberConfig.getUserPassword();
+
+        final HttpGet getRequest = new HttpGet(subscriberUri);
+
+        // add Authorization Header if username and password are present
+        final Optional<String> authHeader = getAuthHeader(userName, userPassword);
+        if (authHeader.isPresent()) {
+            getRequest.addHeader(HttpHeaders.AUTHORIZATION, authHeader.get());
+        } else {
+            LOG.debug("DMaaP MR Subscriber Authentication is disabled as username or password is not present.");
+        }
+
+        try {
+
+            final Pair<Integer, String> responsePair = closeableHttpClient.execute(getRequest, responseHandler());
+            final Integer responseCode = responsePair.getLeft();
+            final String responseBody = responsePair.getRight();
+
+            List<String> fetchedMessages = new LinkedList<>();
+            String responseMessage = responseBody;
+
+            // if messages were published successfully, return successful response
+            if (isSuccessfulResponseCode(responseCode)) {
+                if (responseBody != null) {
+                    fetchedMessages = convertJsonToStringMessages(responseBody);
+                    responseMessage = "Messages Fetched Successfully";
+                } else {
+                    responseMessage = "DMaaP Response Body had no messages";
+                }
+            } else {
+                LOG.error("Unable to fetch messages to DMaaP MR Topic. DMaaP MR unsuccessful Response Code: {}, " +
+                        "DMaaP Response Body: {}", responseCode, responseBody);
+            }
+
+            return createSubscriberResponse(responseCode, responseMessage, fetchedMessages);
+
+        } catch (IOException e) {
+
+            final String errorMessage =
+                    format("IO Exception while fetching messages from DMaaP Topic. Exception %s", e);
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+
+
+    }
+
+    @Override
+    public Date getSubscriberCreationTime() {
+        return new Date(subscriberCreationTime.getTime());
+    }
+
+    @Override
+    public void close() throws Exception {
+        closeableHttpClient.close();
+    }
+}