4 files changed, 393 insertions, 393 deletions
diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java
index aac7fa6..43dc60d 100644
--- a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java
+++ b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiver.java
@@ -1,118 +1,118 @@
-/*
- * ===============================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.cdap.plugins.streaming.dmaap;
-
-import co.cask.cdap.api.data.format.StructuredRecord;
-import co.cask.cdap.api.metrics.Metrics;
-import com.google.common.base.Optional;
-import org.apache.spark.storage.StorageLevel;
-import org.apache.spark.streaming.receiver.Receiver;
-import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-/**
- * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records
- * <p>
- * @author Rajiv Singla . Creation Date: 1/19/2017.
- */
-public class DMaaPMRReceiver extends Receiver<StructuredRecord> {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class);
-    private static final long serialVersionUID = 1L;
-
-    private final DMaaPMRSourcePluginConfig pluginConfig;
-    private final Metrics metrics;
-
-    public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig,
-                           final Metrics metrics) {
-        super(storageLevel);
-        this.pluginConfig = pluginConfig;
-        this.metrics = metrics;
-        LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig);
-    }
-
-    @Override
-    public void onStart() {
-
-        // create DMaaP MR Subscriber
-        final DMaaPMRSubscriber subscriber =
-                DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig));
-
-        // Start a new thread with indefinite loop until receiver is stopped
-        new Thread() {
-            @Override
-            public void run() {
-                while (!isStopped()) {
-                    storeStructuredRecords(subscriber);
-                    try {
-                        final Integer pollingInterval = pluginConfig.getPollingInterval();
-                        LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval);
-                        TimeUnit.MILLISECONDS.sleep(pollingInterval);
-                    } catch (InterruptedException e) {
-                        final String errorMessage = String.format(
-                                "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e);
-                        throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
-                    }
-                }
-            }
-        }.start();
-
-    }
-
-    @Override
-    public void onStop() {
-        LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig);
-    }
-
-    /**
-     * Fetches records from DMaaP MR Subscriber and store them as structured records
-     *
-     * @param subscriber DMaaP MR Subscriber Instance
-     */
-    public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) {
-
-        LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic");
-
-        // Fetch messages from DMaaP MR Topic
-        final Optional<List<String>> subscriberMessagesOptional =
-                DMaaPMRUtils.getSubscriberMessages(subscriber, metrics);
-
-        // store records
-        if (subscriberMessagesOptional.isPresent()) {
-            final List<String> messages = subscriberMessagesOptional.get();
-            for (final String message : messages) {
-                store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message));
-            }
-            LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size());
-        }
-    }
-
-}
+/*
+ * ===============================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.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.api.metrics.Metrics;
+import com.google.common.base.Optional;
+import org.apache.spark.storage.StorageLevel;
+import org.apache.spark.streaming.receiver.Receiver;
+import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records
+ * <p>
+ * @author Rajiv Singla . Creation Date: 1/19/2017.
+ */
+public class DMaaPMRReceiver extends Receiver<StructuredRecord> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRReceiver.class);
+    private static final long serialVersionUID = 1L;
+
+    private final DMaaPMRSourcePluginConfig pluginConfig;
+    private final Metrics metrics;
+
+    public DMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig,
+                           final Metrics metrics) {
+        super(storageLevel);
+        this.pluginConfig = pluginConfig;
+        this.metrics = metrics;
+        LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig);
+    }
+
+    @Override
+    public void onStart() {
+
+        // create DMaaP MR Subscriber
+        final DMaaPMRSubscriber subscriber =
+                DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig));
+
+        // Start a new thread with indefinite loop until receiver is stopped
+        new Thread() {
+            @Override
+            public void run() {
+                while (!isStopped()) {
+                    storeStructuredRecords(subscriber);
+                    try {
+                        final Integer pollingInterval = pluginConfig.getPollingInterval();
+                        LOG.debug("DMaaP MR Receiver sleeping for polling interval: {}", pollingInterval);
+                        TimeUnit.MILLISECONDS.sleep(pollingInterval);
+                    } catch (InterruptedException e) {
+                        final String errorMessage = String.format(
+                                "Interrupted Exception while DMaaP MR Receiver sleeping polling interval: %s", e);
+                        throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+                    }
+                }
+            }
+        }.start();
+
+    }
+
+    @Override
+    public void onStop() {
+        LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig);
+    }
+
+    /**
+     * Fetches records from DMaaP MR Subscriber and store them as structured records
+     *
+     * @param subscriber DMaaP MR Subscriber Instance
+     */
+    public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) {
+
+        LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic");
+
+        // Fetch messages from DMaaP MR Topic
+        final Optional<List<String>> subscriberMessagesOptional =
+                DMaaPMRUtils.getSubscriberMessages(subscriber, metrics);
+
+        // store records
+        if (subscriberMessagesOptional.isPresent()) {
+            final List<String> messages = subscriberMessagesOptional.get();
+            for (final String message : messages) {
+                store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(message));
+            }
+            LOG.debug("Stored DMaaP Subscriber messages as Structured Records. Message count {}", messages.size());
+        }
+    }
+
+}
diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java
index a9ecfea..117c76e 100644
--- a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.java
+++ b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSource.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.cdap.plugins.streaming.dmaap;
-
-import co.cask.cdap.api.annotation.Description;
-import co.cask.cdap.api.annotation.Name;
-import co.cask.cdap.api.annotation.Plugin;
-import co.cask.cdap.api.data.format.StructuredRecord;
-import co.cask.cdap.etl.api.PipelineConfigurer;
-import co.cask.cdap.etl.api.streaming.StreamingContext;
-import co.cask.cdap.etl.api.streaming.StreamingSource;
-import org.apache.spark.storage.StorageLevel;
-import org.apache.spark.streaming.api.java.JavaDStream;
-import org.openecomp.dcae.apod.analytics.cdap.common.utils.ValidationUtils;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.schema.dmaap.DMaaPSourceOutputSchema;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.validator.DMaaPMRSourcePluginConfigValidator;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * DMaaP MR Source Plugin which polls DMaaP MR topic at frequent intervals
- * <p>
- * @author Rajiv Singla . Creation Date: 1/18/2017.
- */
-@Plugin(type = StreamingSource.PLUGIN_TYPE)
-@Name("DMaaPMRSource")
-@Description("Fetches DMaaP MR Messages at regular intervals")
-public class DMaaPMRSource extends StreamingSource<StructuredRecord> {
-
-    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSource.class);
-    private static final long serialVersionUID = 1L;
-
-    private final DMaaPMRSourcePluginConfig pluginConfig;
-
-    public DMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) {
-        LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig);
-        this.pluginConfig = pluginConfig;
-    }
-
-    @Override
-    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
-        ValidationUtils.validateSettings(pluginConfig, new DMaaPMRSourcePluginConfigValidator());
-        pipelineConfigurer.getStageConfigurer().setOutputSchema(DMaaPSourceOutputSchema.getSchema());
-    }
-
-    @Override
-    public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception {
-        return streamingContext.getSparkStreamingContext().receiverStream(
-                new DMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig, streamingContext.getMetrics()));
-    }
-}
+/*
+ * ===============================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.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.annotation.Description;
+import co.cask.cdap.api.annotation.Name;
+import co.cask.cdap.api.annotation.Plugin;
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.etl.api.PipelineConfigurer;
+import co.cask.cdap.etl.api.streaming.StreamingContext;
+import co.cask.cdap.etl.api.streaming.StreamingSource;
+import org.apache.spark.storage.StorageLevel;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.openecomp.dcae.apod.analytics.cdap.common.utils.ValidationUtils;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.schema.dmaap.DMaaPSourceOutputSchema;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.validator.DMaaPMRSourcePluginConfigValidator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * DMaaP MR Source Plugin which polls DMaaP MR topic at frequent intervals
+ * <p>
+ * @author Rajiv Singla . Creation Date: 1/18/2017.
+ */
+@Plugin(type = StreamingSource.PLUGIN_TYPE)
+@Name("DMaaPMRSource")
+@Description("Fetches DMaaP MR Messages at regular intervals")
+public class DMaaPMRSource extends StreamingSource<StructuredRecord> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRSource.class);
+    private static final long serialVersionUID = 1L;
+
+    private final DMaaPMRSourcePluginConfig pluginConfig;
+
+    public DMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) {
+        LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig);
+        this.pluginConfig = pluginConfig;
+    }
+
+    @Override
+    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
+        ValidationUtils.validateSettings(pluginConfig, new DMaaPMRSourcePluginConfigValidator());
+        pipelineConfigurer.getStageConfigurer().setOutputSchema(DMaaPSourceOutputSchema.getSchema());
+    }
+
+    @Override
+    public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception {
+        return streamingContext.getSparkStreamingContext().receiverStream(
+                new DMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig, streamingContext.getMetrics()));
+    }
+}
diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java
index a318406..dc24ca0 100644
--- a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java
+++ b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiver.java
@@ -1,132 +1,132 @@
-/*
- * ===============================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.cdap.plugins.streaming.dmaap;
-
-import co.cask.cdap.api.data.format.StructuredRecord;
-import com.fasterxml.jackson.core.type.TypeReference;
-import org.apache.spark.storage.StorageLevel;
-import org.apache.spark.streaming.receiver.Receiver;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
-import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import static org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.readValue;
-import static org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.writeValueAsString;
-
-/**
- * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records
- * <p>
- * @author Rajiv Singla . Creation Date: 1/19/2017.
- */
-public class MockDMaaPMRReceiver extends Receiver<StructuredRecord> {
-
-    private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRReceiver.class);
-    private static final long serialVersionUID = 1L;
-
-    private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json";
-    private static final TypeReference<List<EventListener>> EVENT_LISTENER_TYPE_REFERENCE =
-            new TypeReference<List<EventListener>>() {
-            };
-
-    private final DMaaPMRSourcePluginConfig pluginConfig;
-
-    public MockDMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig) {
-        super(storageLevel);
-        this.pluginConfig = pluginConfig;
-        LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig);
-    }
-
-    @Override
-    public void onStart() {
-
-        // create DMaaP MR Subscriber
-        final DMaaPMRSubscriber subscriber =
-                DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig));
-        storeStructuredRecords(subscriber);
-
-    }
-
-    @Override
-    public void onStop() {
-        LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig);
-    }
-
-    /**
-     * Fetches records from DMaaP MR Subscriber and store them as structured records
-     *
-     * @param subscriber DMaaP MR Subscriber Instance
-     */
-    public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) {
-
-        LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic");
-        
-        try (InputStream resourceAsStream =
-                     Thread.currentThread().getContextClassLoader().getResourceAsStream(MOCK_MESSAGE_FILE_LOCATION)) {
-
-            if (resourceAsStream == null) {
-                LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION);
-                throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException());
-            }
-
-            List<EventListener> eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE);
-
-            final int totalMessageCount = eventListeners.size();
-            LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount);
-
-            int i = 1;
-            for (EventListener eventListener : eventListeners) {
-                if (isStopped()) {
-                    return;
-                }
-                final String eventListenerString = writeValueAsString(eventListener);
-                LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount);
-                store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(eventListenerString));
-                i++;
-                try {
-                    TimeUnit.MILLISECONDS.sleep(pluginConfig.getPollingInterval());
-                } catch (InterruptedException e) {
-                    LOG.error("Error while sleeping");
-                    throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e);
-                }
-
-            }
-
-            LOG.debug("Finished writing mock messages to CDAP Stream");
-
-        } catch (IOException e) {
-            LOG.error("Error while parsing json file");
-            throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e);
-        }
-    }
-
-}
+/*
+ * ===============================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.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.spark.storage.StorageLevel;
+import org.apache.spark.streaming.receiver.Receiver;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.CDAPPluginUtils;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.utils.DMaaPSourceConfigMapper;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.openecomp.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.openecomp.dcae.apod.analytics.model.domain.cef.EventListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.readValue;
+import static org.openecomp.dcae.apod.analytics.model.util.AnalyticsModelJsonUtils.writeValueAsString;
+
+/**
+ * DMaaP MR Receiver which calls DMaaP MR Topic and stores structured records
+ * <p>
+ * @author Rajiv Singla . Creation Date: 1/19/2017.
+ */
+public class MockDMaaPMRReceiver extends Receiver<StructuredRecord> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRReceiver.class);
+    private static final long serialVersionUID = 1L;
+
+    private static final String MOCK_MESSAGE_FILE_LOCATION = "ves_mock_messages.json";
+    private static final TypeReference<List<EventListener>> EVENT_LISTENER_TYPE_REFERENCE =
+            new TypeReference<List<EventListener>>() {
+            };
+
+    private final DMaaPMRSourcePluginConfig pluginConfig;
+
+    public MockDMaaPMRReceiver(final StorageLevel storageLevel, final DMaaPMRSourcePluginConfig pluginConfig) {
+        super(storageLevel);
+        this.pluginConfig = pluginConfig;
+        LOG.debug("Created DMaaP MR Receiver instance with plugin Config: {}", pluginConfig);
+    }
+
+    @Override
+    public void onStart() {
+
+        // create DMaaP MR Subscriber
+        final DMaaPMRSubscriber subscriber =
+                DMaaPMRFactory.create().createSubscriber(DMaaPSourceConfigMapper.map(pluginConfig));
+        storeStructuredRecords(subscriber);
+
+    }
+
+    @Override
+    public void onStop() {
+        LOG.debug("Stopping DMaaP MR Receiver with plugin config: {}", pluginConfig);
+    }
+
+    /**
+     * Fetches records from DMaaP MR Subscriber and store them as structured records
+     *
+     * @param subscriber DMaaP MR Subscriber Instance
+     */
+    public void storeStructuredRecords(final DMaaPMRSubscriber subscriber) {
+
+        LOG.debug("DMaaP MR Receiver start fetching messages from DMaaP MR Topic");
+        
+        try (InputStream resourceAsStream =
+                     Thread.currentThread().getContextClassLoader().getResourceAsStream(MOCK_MESSAGE_FILE_LOCATION)) {
+
+            if (resourceAsStream == null) {
+                LOG.error("Unable to find file at location: {}", MOCK_MESSAGE_FILE_LOCATION);
+                throw new DCAEAnalyticsRuntimeException("Unable to find file", LOG, new FileNotFoundException());
+            }
+
+            List<EventListener> eventListeners = readValue(resourceAsStream, EVENT_LISTENER_TYPE_REFERENCE);
+
+            final int totalMessageCount = eventListeners.size();
+            LOG.debug("Mock message count to be written to cdap stream: ()", totalMessageCount);
+
+            int i = 1;
+            for (EventListener eventListener : eventListeners) {
+                if (isStopped()) {
+                    return;
+                }
+                final String eventListenerString = writeValueAsString(eventListener);
+                LOG.debug("=======>> Writing message to cdap stream no: {} of {}", i, totalMessageCount);
+                store(CDAPPluginUtils.createDMaaPMRResponseStructuredRecord(eventListenerString));
+                i++;
+                try {
+                    TimeUnit.MILLISECONDS.sleep(pluginConfig.getPollingInterval());
+                } catch (InterruptedException e) {
+                    LOG.error("Error while sleeping");
+                    throw new DCAEAnalyticsRuntimeException("Error while sleeping", LOG, e);
+                }
+
+            }
+
+            LOG.debug("Finished writing mock messages to CDAP Stream");
+
+        } catch (IOException e) {
+            LOG.error("Error while parsing json file");
+            throw new DCAEAnalyticsRuntimeException("Error while parsing mock json file", LOG, e);
+        }
+    }
+
+}
diff --git a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java
index e0be12f..e058fab 100644
--- a/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java
+++ b/dcae-analytics-cdap-plugins/src/main/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSource.java
@@ -1,73 +1,73 @@
-/*
- * ===============================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.cdap.plugins.streaming.dmaap;
-
-import co.cask.cdap.api.annotation.Description;
-import co.cask.cdap.api.annotation.Name;
-import co.cask.cdap.api.annotation.Plugin;
-import co.cask.cdap.api.data.format.StructuredRecord;
-import co.cask.cdap.etl.api.PipelineConfigurer;
-import co.cask.cdap.etl.api.streaming.StreamingContext;
-import co.cask.cdap.etl.api.streaming.StreamingSource;
-import org.apache.spark.storage.StorageLevel;
-import org.apache.spark.streaming.api.java.JavaDStream;
-import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
-import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A mock implementation of DMaaP MR Receiver which sends mock ves messages
- * <p>
- * @author Rajiv Singla . Creation Date: 2/15/2017.
- */
-@Plugin(type = StreamingSource.PLUGIN_TYPE)
-@Name("MockDMaaPMRSource")
-@Description("Fetches DMaaP MR Messages at regular intervals")
-public class MockDMaaPMRSource extends StreamingSource<StructuredRecord> {
-
-    private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRSource.class);
-    private static final long serialVersionUID = 1L;
-
-    private final DMaaPMRSourcePluginConfig pluginConfig;
-
-    public MockDMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) {
-        LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig);
-        this.pluginConfig = pluginConfig;
-    }
-
-    @Override
-    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
-        final Integer pollingInterval = pluginConfig.getPollingInterval();
-        if (pollingInterval == null) {
-            final String errorMessage = "Polling Interval field must be present";
-            throw new CDAPSettingsException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
-        } else {
-            LOG.info("Mock Message will be send every ms: {}", pollingInterval);
-        }
-    }
-
-    @Override
-    public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception {
-        return streamingContext.getSparkStreamingContext().receiverStream(
-                new MockDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig));
-    }
-}
+/*
+ * ===============================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.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.annotation.Description;
+import co.cask.cdap.api.annotation.Name;
+import co.cask.cdap.api.annotation.Plugin;
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.etl.api.PipelineConfigurer;
+import co.cask.cdap.etl.api.streaming.StreamingContext;
+import co.cask.cdap.etl.api.streaming.StreamingSource;
+import org.apache.spark.storage.StorageLevel;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A mock implementation of DMaaP MR Receiver which sends mock ves messages
+ * <p>
+ * @author Rajiv Singla . Creation Date: 2/15/2017.
+ */
+@Plugin(type = StreamingSource.PLUGIN_TYPE)
+@Name("MockDMaaPMRSource")
+@Description("Fetches DMaaP MR Messages at regular intervals")
+public class MockDMaaPMRSource extends StreamingSource<StructuredRecord> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MockDMaaPMRSource.class);
+    private static final long serialVersionUID = 1L;
+
+    private final DMaaPMRSourcePluginConfig pluginConfig;
+
+    public MockDMaaPMRSource(final DMaaPMRSourcePluginConfig pluginConfig) {
+        LOG.debug("Creating DMaaP MR Source plugin with plugin Config: {}", pluginConfig);
+        this.pluginConfig = pluginConfig;
+    }
+
+    @Override
+    public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
+        final Integer pollingInterval = pluginConfig.getPollingInterval();
+        if (pollingInterval == null) {
+            final String errorMessage = "Polling Interval field must be present";
+            throw new CDAPSettingsException(errorMessage, LOG, new IllegalArgumentException(errorMessage));
+        } else {
+            LOG.info("Mock Message will be send every ms: {}", pollingInterval);
+        }
+    }
+
+    @Override
+    public JavaDStream<StructuredRecord> getStream(final StreamingContext streamingContext) throws Exception {
+        return streamingContext.getSparkStreamingContext().receiverStream(
+                new MockDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), pluginConfig));
+    }
+}