Initial TCA commit into DCAEGEN2

Change-Id: I5f7f8af2a00419854cafc34b79277df60d1af095
Issue-ID: DCAEGEN2-53
Signed-off-by: an4828 <nekrassov@att.com>

18 files changed, 2027 insertions, 0 deletions

diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPComponentsConstants.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPComponentsConstants.java
new file mode 100644
index 0000000..d00bc1b
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPComponentsConstants.java
@@ -0,0 +1,200 @@
+/*
+ * ===============================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.common;
+
+/**
+ *  Contains static constant variable names and values of all DCAE CDAP Components for
+ *  e.g. app names, app descriptions, streams, datasets, flows, flowlets, workers, outputs etc.
+ *
+ *  <p>
+ *      A strict naming convention must be followed for variable names for CDAP Components names for proper
+ *      identification of CDAP Component variable purpose and function. A custom configuration settings can be
+ *      generated for app
+ *      deployment purposes based on variable naming conventions.
+ *
+ *      The variable names should have 4 parts separated by an underscore:
+ *          <ul>
+ *              <li>Name of the DCAE sub module (e.g. TCA) to which variable is applicable</li>
+ *              <li>Information about variable name:
+ *                  <ul>
+ *                      <li>FIXED - if variable value is fixed and cannot be changed</li>
+ *                      <li>DEFAULT - if variable name is default name and can be changed by cdap settings file
+ *                                    when application is created</li>
+ *                  </ul>
+ *              </li>
+ *              <li>Actual Descriptive name about the CDAP component (may contain underscrores) </li>
+ *              <li>CDAP component type e.g STREAM, DATASET, APP, FLOW, FLOWLET, OUTPUT</li>
+ *          </ul>
+ *
+ *  <p>e.g TCA_DEFAULT_DMAAP_INPUT_STREAM</p>
+ *
+ *  <p><strong>RegEx Format (DCAE MODULE NAME)_(FIXED|DEFAULT)_(VARIABLE NAME)_(CDAP COMPONENT TYPE)</strong></p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/24/2016.
+ */
+public abstract class CDAPComponentsConstants {
+
+    // ===============  Common Module Constants ==================== //
+
+    /**
+     * Default DCAE App Name. It should be overriden by sub modules
+     */
+    public static final String COMMON_DEFAULT_DCAE_CDAP_NAME_APP = "DCAE_ANALYTICS_GENERIC_APP";
+
+    /**
+     * Default DCAE App Description. It should be overriden by sub modules
+     */
+    public static final String COMMON_DEFAULT_DCAE_CDAP_DESCRIPTION_APP = "DCAE ANALYTICS GENERIC APP DESCRIPTION";
+
+    // ===============  TCA Module Constants ==================== //
+
+    /**
+     * Default TCA application name if application name is not provided from startup configuration
+     */
+    public static final String TCA_DEFAULT_NAME_APP = "dcae-tca";
+
+    /**
+     * Default TCA application description if not provided from startup configuration
+     */
+    public static final String TCA_DEFAULT_DESCRIPTION_APP = "DCAE Analytics Threshold Crossing Alert Application";
+
+    /**
+     * Default TCA DMaaP Subscriber output stream name if not provided from startup configuration
+     */
+    public static final String TCA_DEFAULT_SUBSCRIBER_OUTPUT_NAME_STREAM = "TCASubscriberOutputStream";
+
+
+    /**
+     * Fixed TCA DMaaP Subscriber output stream description
+     */
+    public static final String TCA_FIXED_SUBSCRIBER_OUTPUT_DESCRIPTION_STREAM =
+            "Stream which contains all message from VES Collector DMaaP MR topic";
+
+
+    /**
+     * Fixed Name of TCA DMaaP Subscriber Worker - which will be fetching DMaaP Messages posting them to CDAP stream
+     */
+    public static final String TCA_FIXED_DMAAP_SUBSCRIBER_WORKER = "TCADMaaPMRSubscriberWorker";
+
+    /**
+     * Fixed Description of TCA DMaaP Subscriber Worker
+     */
+    public static final String TCA_FIXED_DMAAP_SUBSCRIBER_DESCRIPTION_WORKER =
+            "Fetches messages from DMaaP MR Topic at frequent intervals and writes them to a CDAP stream";
+
+    /**
+     * Fixed Name of TCA DMaaP Publisher Worker - which will be publishing messages to DMaaP MR
+     */
+    public static final String TCA_FIXED_DMAAP_PUBLISHER_WORKER = "TCADMaaPMRPublisherWorker";
+
+    /**
+     * Fixed Description of TCA DMaaP Publisher Worker
+     */
+    public static final String TCA_FIXED_DMAAP_PUBLISHER_DESCRIPTION_WORKER =
+            "Polls TCA Alerts Table at frequent intervals for new alerts and publishes them to DMaaP MR Topic";
+
+    /**
+     * Fixed name for TCA VES Collector Messages Processing Flow
+     */
+    public static final String TCA_FIXED_VES_COLLECTOR_NAME_FLOW = "TCAVESCollectorFlow";
+
+
+    /**
+     * Fixed description for TCA VES Collector Messages Processing Flow
+     */
+    public static final String TCA_FIXED_VES_COLLECTOR_DESCRIPTION_FLOW = "Flow performs TCA on VES Collector Messages";
+
+
+    /**
+     * Fixed Name for TCA VES Message Router Flowlet
+     */
+    public static final String TCA_FIXED_VES_MESSAGE_ROUTER_NAME_FLOWLET = "TCAVESMessageRouterFlowlet";
+
+    /**
+     * Fixed Description for TCA VES Message Router Flowlet
+     */
+    public static final String TCA_FIXED_VES_MESSAGE_ROUTER_DESCRIPTION_FLOWLET =
+            "Routes message received from TCA VES Collector to TCA Threshold Calculator Flowlet instances";
+
+    /**
+     * Fixed TCA VES Message Router Flowlet Output
+     */
+    public static final String TCA_FIXED_VES_MESSAGE_ROUTER_OUTPUT = "TCAVESMessageRouterFlowlet";
+
+    /**
+     * Fixed Name for TCA VES Message Policy Violated Threshold Calculator Flowlet
+     */
+    public static final String TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_NAME_FLOWLET =
+            "TCAVESThresholdViolationCalculatorFlowlet";
+
+    /**
+     * Fixed Description for TCA VES Message Policy Violated Threshold Calculator Flowlet
+     */
+    public static final String TCA_FIXED_VES_THRESHOLD_VIOLATION_CALCULATOR_DESCRIPTION_FLOWLET =
+            "Applies TCA Policy Thresholds to VES Message and determined if any message violated TCA Policy thresholds";
+
+    /**
+     * Fixed Name for TCA VES Message Policy Violated Threshold Calculator Output
+     */
+    public static final String TCA_FIXED_VES_TCA_CALCULATOR_NAME_OUTPUT = "TCAThresholdViolationCalculatorOutput";
+
+
+    /**
+     * Fixed Name for TCA VES Alerts Sink Flowlet
+     */
+    public static final String TCA_FIXED_VES_ALERTS_SINK_NAME_FLOWLET = "TCAVESAlertsSinkFlowlet";
+
+    /**
+     * Fixed Description for TCA VES Alerts Sink Flowlet
+     */
+    public static final String TCA_FIXED_VES_ALERTS_SINK_DESCRIPTION_FLOWLET =
+            "Saves messages which violated TCA Policy in a data set";
+
+
+    /**
+     * Default Name for TCA VES Message status table which contain status of all messages processed by TCA
+     */
+    public static final String TCA_DEFAULT_VES_MESSAGE_STATUS_NAME_TABLE = "TCAVESMessageStatusTable";
+
+
+    /**
+     * Fixed Description for TCA VES Message status table which contain status of all messages processed by TCA
+     */
+    public static final String TCA_FIXED_VES_MESSAGE_STATUS_DESCRIPTION_TABLE =
+            "Store processing information about all incoming TCA VES Messages";
+
+    /**
+     * Default Name for TCA VES Alerts table which contains alerts that can be send to downstream systems
+     */
+    public static final String TCA_DEFAULT_VES_ALERTS_NAME_TABLE = "TCAVESAlertsTable";
+
+    /**
+     * Fixed Description for TCA VES Alerts table which contains alerts that can be send to downstream systems
+     */
+    public static final String TCA_DEFAULT_VES_ALERTS_DESCRIPTION_TABLE =
+            "Stores alert messages that need to be DMaaP";
+
+
+    private CDAPComponentsConstants() {
+
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPMetricsConstants.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPMetricsConstants.java
new file mode 100644
index 0000000..f2e0e73
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPMetricsConstants.java
@@ -0,0 +1,119 @@
+/*
+ * ===============================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.common;
+
+/**
+ * Contains all metrics names used for DCAE CDAP modules
+ *
+ * <p>
+ *     Format should be (ModuleName)_(Description of metrics)_METRIC e.g. TCA_WORKER_FAILED_ATTEMPTS_METRIC
+ * </p>
+ *
+ * @author Rajiv Singla . Creation Date: 10/25/2016.
+ */
+public abstract class CDAPMetricsConstants {
+
+    /**
+     * Metric captures count of all responses received from DMaaP MR Subscriber Topic
+     */
+    public static final String DMAAP_MR_SUBSCRIBER_ALL_RESPONSES_COUNT_METRIC =
+            "dmaap.subscriber.fetch.all_responses";
+
+    /**
+     * Metric captures count of responses from DMaaP MR Subscriber Topic which does not have 200 HTTP Response code.
+     * This can be due to DMaaP topic being down or any internal server errors etc.
+     */
+    public static final String DMAAP_MR_SUBSCRIBER_UNSUCCESSFUL_RESPONSES_METRIC =
+            "dmaap.subscriber.fetch.unsuccessful";
+
+    /**
+     * Metric that counts the number of successful (200 HTTP Response Code) calls to DMaaP which did not had empty
+     * messages
+     */
+    public static final String DMAAP_MR_SUBSCRIBER_RESPONSES_WITH_NO_MESSAGES_METRIC =
+            "dmaap.subscriber.fetch.no_message";
+
+    /**
+     * Metric to count total number of message processed by DMaaP MR subscriber
+     */
+    public static final String DMAAP_MR_SUBSCRIBER_TOTAL_MESSAGES_PROCESSED_METRIC = "dmaap.subscriber.message.count";
+
+
+    /**
+     * Metric to record time taken in ms by subscriber in its most recent call to fetch DMaaP MR messages
+     */
+    public static final String DMAAP_MR_SUBSCRIBER_RESPONSE_TIME_MS_METRIC = "dmaap.subscriber.fetch.response_time";
+
+    /**
+     * Metric captures the count of number of times DMaaP MR Subscriber was unable to write to DMaaP Stream due
+     * some CDAP error while writing to stream. This should ideally never happen assuming we have enough space
+     * on CDAP machine and CDAP process is functioning normally
+     */
+    public static final String TCA_SUBSCRIBER_FAILURE_TO_WRITE_TO_STREAM_METRIC = "tca.subscriber.stream.writing.error";
+
+
+    /**
+     * Metric captures number of VES messages that are not applicable as per TCA Policy
+     */
+    public static final String TCA_VES_INAPPLICABLE_MESSAGES_METRIC = "tca.ves.calculator.inapplicable";
+
+    /**
+     * Metric captures number of VES messages that are applicable as per TCA Policy but don't violate any thresholds
+     */
+    public static final String TCA_VES_COMPLIANT_MESSAGES_METRIC = "tca.ves.calculator.compliant";
+
+    /**
+     * Metrics captures number of VES messages that are applicable as per TCA Policy and does violate thresholds and
+     * will likely cause an alert
+     */
+    public static final String TCA_VES_NON_COMPLIANT_MESSAGES_METRIC = "tca.ves.calculator.non_compliant";
+
+
+    /**
+     * Metric that counts the number of publisher look ups in alerts table which resulted in 0 new alerts
+     */
+    public static final String TCA_PUBLISHER_NO_NEW_ALERTS_LOOKUP_METRIC = "tca.publisher.lookup.no_message";
+
+    /**
+     * Metric that counts the number of new alerts found by the publisher in alerts table
+     */
+    public static final String TCA_PUBLISHER_NEW_ALERTS_METRIC = "tca.publisher.lookup.new_messages";
+
+    /**
+     * Metric that counts the number of alerts deleted by publisher in alerts table
+     */
+    public static final String TCA_PUBLISHER_DELETED_ALERTS_METRIC = "tca.publisher.deleted.alerts";
+
+    /**
+     * Metric that counts the number of publisher calls to DMaaP which resulted in successful response code
+     */
+    public static final String TCA_PUBLISHER_SUCCESSFUL_DMAAP_RESPONSE_METRIC = "tca.publisher.publish.successful";
+
+    /**
+     * Metric that counts the number of publisher calls to DMaaP which resulted in unsuccessful response code
+     */
+    public static final String TCA_PUBLISHER_UNSUCCESSFUL_DMAAP_RESPONSE_METRIC = "tca.publisher.publish.unsuccessful";
+
+    private CDAPMetricsConstants() {
+
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPPluginConstants.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPPluginConstants.java
new file mode 100644
index 0000000..c3e61c0
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/CDAPPluginConstants.java
@@ -0,0 +1,71 @@
+/*
+ * ===============================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.common;
+
+/**
+ * <p>
+ *      Contains CDAP Plugin Constants
+ * </p>
+ * @author Rajiv Singla . Creation Date: 1/17/2017.
+ */
+public abstract class CDAPPluginConstants {
+
+    /**
+     * Common Reference Name property name and description used to create an external Dataset for metadata, lineage
+     * purposes
+     */
+    public static class Reference {
+
+        public static final String REFERENCE_NAME = "referenceName";
+        public static final String REFERENCE_NAME_DESCRIPTION =
+                "This will be used to uniquely identify this source/sink for lineage, annotating metadata, etc.";
+
+        private Reference() {
+            // private constructor
+        }
+
+    }
+
+    /**
+     * Contains fields for DMaaP MR Sink Map Reduce Hadoop Configuration
+     */
+    public static class DMaaPMRSinkHadoopConfigFields {
+
+        public static final String HOST_NAME = "dmaap.mr.sink.hostName";
+        public static final String PORT_NUMBER = "dmaap.mr.sink.portNumber";
+        public static final String TOPIC_NAME = "dmaap.mr.sink.topicName";
+        public static final String PROTOCOL = "dmaap.mr.sink.protocol";
+        public static final String USER_NAME = "dmaap.mr.sink.userName";
+        public static final String USER_PASS = "dmaap.mr.sink.userPassword";
+        public static final String CONTENT_TYPE = "dmaap.mr.sink.contentType";
+        public static final String MAX_BATCH_SIZE = "dmaap.mr.sink.maxBatchSize";
+        public static final String MAX_RECOVER_QUEUE_SIZE = "dmaap.mr.sink.maxRecoveryQueueSize";
+
+        private DMaaPMRSinkHadoopConfigFields() {
+            // private constructor
+        }
+    }
+
+
+    private CDAPPluginConstants() {
+        // private constructor
+    }
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/exception/CDAPSettingsException.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/exception/CDAPSettingsException.java
new file mode 100644
index 0000000..2e6d90f
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/exception/CDAPSettingsException.java
@@ -0,0 +1,51 @@
+/*
+ * ===============================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.common.exception;
+
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.slf4j.Logger;
+
+/**
+ * Runtime exception which signals that DCAE CDAP App settings (e.g. app config, preferences) validation failed
+ *
+ * @author Rajiv Singla . Creation Date: 10/24/2016.
+ */
+public class CDAPSettingsException extends DCAEAnalyticsRuntimeException {
+
+    /**
+     * @param message - Error Message for Exception
+     * @param cause   - Actual Exception which caused {@link DCAEAnalyticsRuntimeException}
+     */
+    public CDAPSettingsException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    /**
+     * Creates and logs the DCAE App Config Exception to given logger
+     *
+     * @param message - Error Message for Exception and logging
+     * @param logger  - Logger used for logging exception
+     * @param cause   - Actual exception which caused
+     */
+    public CDAPSettingsException(String message, Logger logger, Throwable cause) {
+        super(message, logger, cause);
+    }
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCACalculatorMessageType.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCACalculatorMessageType.java
new file mode 100644
index 0000000..957fc2e
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCACalculatorMessageType.java
@@ -0,0 +1,43 @@
+/*
+ * ===============================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.common.persistance.tca;
+
+/**
+ * TCA Calculator applies TCA Policy to incoming VES messages and classifies them as per this enum
+ *
+ * @author Rajiv Singla . Creation Date: 11/15/2016.
+ */
+public enum TCACalculatorMessageType {
+
+    /**
+     * VES messages that are not applicable as per TCA Policy
+     */
+    INAPPLICABLE,
+    /**
+     * VES messages that are applicable as per TCA Policy but don't violate any thresholds
+     */
+    COMPLIANT,
+    /**
+     * VES messages that are applicable as per TCA Policy and also in violation of TCA Policy thresholds
+     */
+    NON_COMPLIANT;
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusEntity.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusEntity.java
new file mode 100644
index 0000000..c529e55
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusEntity.java
@@ -0,0 +1,555 @@
+/*
+ * ===============================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.common.persistance.tca;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+import org.openecomp.dcae.apod.analytics.model.domain.cef.EventSeverity;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Direction;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.TCAPolicy;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFJsonProcessor;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyDomainFilter;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyFunctionalRoleFilter;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyThresholdsProcessor;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * TCA Message Status is an Entity which is used to persist TCA VES Message status information in Message Status Table
+ *
+ * @author Rajiv Singla . Creation Date: 11/16/2016.
+ */
+public class TCAMessageStatusEntity implements Writable, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private long creationTS;
+    private int instanceId;
+    private String messageType;
+    private String vesMessage;
+    private String domain;
+    private String functionalRole;
+    private String thresholdPath;
+    private String thresholdSeverity;
+    private String thresholdDirection;
+    private Long thresholdValue;
+    private String jsonProcessorStatus;
+    private String jsonProcessorMessage;
+    private String domainFilterStatus;
+    private String domainFilterMessage;
+    private String functionalRoleFilterStatus;
+    private String functionalRoleFilterMessage;
+    private String thresholdCalculatorStatus;
+    private String thresholdCalculatorMessage;
+    private String alertMessage;
+
+    /**
+     * No Arg constructor required for Jackson Json Serialization / Deserialization
+     */
+    public TCAMessageStatusEntity() {
+        // no argument constructor required for json serialization / deserialization
+    }
+
+    /**
+     * Create new Instance of {@link TCAMessageStatusEntity}
+     *
+     * @param creationTS creation Timestamp
+     * @param instanceId CDAP flowlet instance ID
+     * @param messageType {@link TCACalculatorMessageType}
+     * @param vesMessage incoming VES message from collector
+     * @param domain VES message domain if present
+     * @param functionalRole VES message functional role if present
+     */
+    public TCAMessageStatusEntity(final long creationTS, final int instanceId, final String messageType,
+                                  final String vesMessage, final String domain, final String functionalRole) {
+        this(creationTS, instanceId, messageType, vesMessage, domain, functionalRole, null, null, null, null,
+                null, null, null, null, null, null, null, null, null);
+    }
+
+
+    /**
+     * Create new Instance of {@link TCAMessageStatusEntity}
+     *
+     * @param creationTS creation Timestamp
+     * @param instanceId CDAP flowlet instance ID
+     * @param messageType {@link TCACalculatorMessageType}
+     * @param vesMessage incoming VES message from collector
+     * @param domain VES message domain if present
+     * @param functionalRole VES message functional role if present
+     * @param thresholdPath Violated threshold path
+     * @param thresholdSeverity Violated threshold Severity if any
+     * @param thresholdDirection Violated threshold Direction if any
+     * @param thresholdValue Violated threshold value if any
+     * @param jsonProcessorStatus {@link TCACEFJsonProcessor} status
+     * @param jsonProcessorMessage {@link TCACEFJsonProcessor} message
+     * @param domainFilterStatus {@link TCACEFPolicyDomainFilter} status
+     * @param domainFilterMessage {@link TCACEFPolicyDomainFilter} message
+     * @param functionalRoleFilterStatus
+     * {@link TCACEFPolicyFunctionalRoleFilter} status
+     * @param functionalRoleFilterMessage
+     * {@link TCACEFPolicyFunctionalRoleFilter} message
+     * @param thresholdCalculatorStatus
+     * {@link TCACEFPolicyThresholdsProcessor} status
+     * @param thresholdCalculatorMessage
+     * {@link TCACEFPolicyThresholdsProcessor} message
+     * @param alertMessage alert message that will be sent out in case of threshold violation
+     */
+    public TCAMessageStatusEntity(long creationTS, int instanceId, String messageType, String vesMessage,
+                                  String domain, String functionalRole,
+                                  String thresholdPath, String thresholdSeverity, String thresholdDirection,
+                                  Long thresholdValue,
+                                  String jsonProcessorStatus, String jsonProcessorMessage,
+                                  String domainFilterStatus, String domainFilterMessage,
+                                  String functionalRoleFilterStatus, String functionalRoleFilterMessage,
+                                  String thresholdCalculatorStatus, String thresholdCalculatorMessage,
+                                  String alertMessage) {
+        this.creationTS = creationTS;
+        this.instanceId = instanceId;
+        this.messageType = messageType;
+        this.vesMessage = vesMessage;
+        this.domain = domain;
+        this.functionalRole = functionalRole;
+        this.thresholdPath = thresholdPath;
+        this.thresholdSeverity = thresholdSeverity;
+        this.thresholdDirection = thresholdDirection;
+        this.thresholdValue = thresholdValue;
+        this.jsonProcessorStatus = jsonProcessorStatus;
+        this.jsonProcessorMessage = jsonProcessorMessage;
+        this.domainFilterStatus = domainFilterStatus;
+        this.domainFilterMessage = domainFilterMessage;
+        this.functionalRoleFilterStatus = functionalRoleFilterStatus;
+        this.functionalRoleFilterMessage = functionalRoleFilterMessage;
+        this.thresholdCalculatorStatus = thresholdCalculatorStatus;
+        this.thresholdCalculatorMessage = thresholdCalculatorMessage;
+        this.alertMessage = alertMessage;
+    }
+
+    /**
+     * Provides Creation Timestamp
+     *
+     * @return creation timestamp long value
+     */
+    public long getCreationTS() {
+        return creationTS;
+    }
+
+    /**
+     * Sets Creations Timestamp
+     *
+     * @param creationTS creation timestamp long value
+     */
+    public void setCreationTS(long creationTS) {
+        this.creationTS = creationTS;
+    }
+
+
+    /**
+     * Provides CDAP Flowlet instance ID
+     *
+     * @return cdap flowlet instance ID
+     */
+    public int getInstanceId() {
+        return instanceId;
+    }
+
+    /**
+     * Sets CDAP Flowlet instance ID
+     *
+     * @param instanceId flowlet instance ID
+     */
+    public void setInstanceId(int instanceId) {
+        this.instanceId = instanceId;
+    }
+
+    /**
+     * Provides Message Calculator Type {@link TCACalculatorMessageType}
+     *
+     * @return calculator message type
+     */
+    public String getMessageType() {
+        return messageType;
+    }
+
+    /**
+     * Sets Calculator message Type {@link TCACalculatorMessageType}
+     *
+     * @param messageType calculator message type
+     */
+    public void setMessageType(String messageType) {
+        this.messageType = messageType;
+    }
+
+    /**
+     * Provides incoming VES Message
+     *
+     * @return ves message
+     */
+    public String getVesMessage() {
+        return vesMessage;
+    }
+
+    /**
+     * Set new value for VES message
+     *
+     * @param vesMessage ves message
+     */
+    public void setVesMessage(String vesMessage) {
+        this.vesMessage = vesMessage;
+    }
+
+    /**
+     * Provides VES message Domain
+     *
+     * @return ves message domain
+     */
+    public String getDomain() {
+        return domain;
+    }
+
+    /**
+     * Sets VES Message Domain
+     *
+     * @param domain ves message domain
+     */
+    public void setDomain(String domain) {
+        this.domain = domain;
+    }
+
+    /**
+     * Provides VES Message Functional Role
+     *
+     * @return ves message functional role
+     */
+    public String getFunctionalRole() {
+        return functionalRole;
+    }
+
+    /**
+     * Sets VES Message Functional Role
+     *
+     * @param functionalRole ves message Functional Role
+     */
+    public void setFunctionalRole(String functionalRole) {
+        this.functionalRole = functionalRole;
+    }
+
+    /**
+     * Violated Threshold Path as extracted from {@link TCAPolicy}
+     *
+     * @return violated threshold path
+     */
+    public String getThresholdPath() {
+        return thresholdPath;
+    }
+
+    /**
+     * Sets value for Violated Threshold Path
+     *
+     * @param thresholdPath violated threshold path
+     */
+    public void setThresholdPath(String thresholdPath) {
+        this.thresholdPath = thresholdPath;
+    }
+
+    /**
+     * Violated threshold {@link EventSeverity}
+     *
+     * @return event severity
+     */
+    public String getThresholdSeverity() {
+        return thresholdSeverity;
+    }
+
+    /**
+     * Violated Threshold Severity
+     *
+     * @param thresholdSeverity violated threshold severity
+     */
+    public void setThresholdSeverity(String thresholdSeverity) {
+        this.thresholdSeverity = thresholdSeverity;
+    }
+
+    /**
+     * Violated Threshold {@link Direction}
+     *
+     * @return violated threshold Direction
+     */
+    public String getThresholdDirection() {
+        return thresholdDirection;
+    }
+
+    /**
+     * Sets Violated Threshold Direction
+     *
+     * @param thresholdDirection violated threshold direction
+     */
+    public void setThresholdDirection(String thresholdDirection) {
+        this.thresholdDirection = thresholdDirection;
+    }
+
+    /**
+     * Provides Violated Threshold Value
+     *
+     * @return violated Threshold value
+     */
+    public Long getThresholdValue() {
+        return thresholdValue;
+    }
+
+    /**
+     * Sets Violated Threshold Value
+     *
+     * @param thresholdValue violated threshold value
+     */
+    public void setThresholdValue(Long thresholdValue) {
+        this.thresholdValue = thresholdValue;
+    }
+
+    /**
+     * Provides {@link TCACEFJsonProcessor} status
+     *
+     * @return json processor status
+     */
+    public String getJsonProcessorStatus() {
+        return jsonProcessorStatus;
+    }
+
+    /**
+     * Sets Json Processor status
+     *
+     * @param jsonProcessorStatus json processor status
+     */
+    public void setJsonProcessorStatus(String jsonProcessorStatus) {
+        this.jsonProcessorStatus = jsonProcessorStatus;
+    }
+
+    /**
+     * Provides {@link TCACEFJsonProcessor} message
+     *
+     * @return json processor message
+     */
+    public String getJsonProcessorMessage() {
+        return jsonProcessorMessage;
+    }
+
+    /**
+     * Sets Json Processor Message
+     *
+     * @param jsonProcessorMessage json processor message
+     */
+    public void setJsonProcessorMessage(String jsonProcessorMessage) {
+        this.jsonProcessorMessage = jsonProcessorMessage;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyDomainFilter} status
+     *
+     * @return domain filter status
+     */
+    public String getDomainFilterStatus() {
+        return domainFilterStatus;
+    }
+
+    /**
+     * Sets Domain Filter status
+     *
+     * @param domainFilterStatus domain filter status
+     */
+    public void setDomainFilterStatus(String domainFilterStatus) {
+        this.domainFilterStatus = domainFilterStatus;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyDomainFilter} message
+     *
+     * @return domain filter message
+     */
+    public String getDomainFilterMessage() {
+        return domainFilterMessage;
+    }
+
+    /**
+     * Sets Domain filter message
+     *
+     * @param domainFilterMessage domain filter message
+     */
+    public void setDomainFilterMessage(String domainFilterMessage) {
+        this.domainFilterMessage = domainFilterMessage;
+    }
+
+    public String getFunctionalRoleFilterStatus() {
+        return functionalRoleFilterStatus;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyFunctionalRoleFilter} status
+     *
+     * @param functionalRoleFilterStatus functional Role filter status
+     */
+    public void setFunctionalRoleFilterStatus(String functionalRoleFilterStatus) {
+        this.functionalRoleFilterStatus = functionalRoleFilterStatus;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyFunctionalRoleFilter} message
+     *
+     * @return functional role filter message
+     */
+    public String getFunctionalRoleFilterMessage() {
+        return functionalRoleFilterMessage;
+    }
+
+    /**
+     * Sets Functional Role filter message
+     *
+     * @param functionalRoleFilterMessage functional role filter message
+     */
+    public void setFunctionalRoleFilterMessage(String functionalRoleFilterMessage) {
+        this.functionalRoleFilterMessage = functionalRoleFilterMessage;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyThresholdsProcessor} status
+     *
+     * @return threshold processor status
+     */
+    public String getThresholdCalculatorStatus() {
+        return thresholdCalculatorStatus;
+    }
+
+    /**
+     * Sets threshold calculator status
+     *
+     * @param thresholdCalculatorStatus threshold calculator status
+     */
+    public void setThresholdCalculatorStatus(String thresholdCalculatorStatus) {
+        this.thresholdCalculatorStatus = thresholdCalculatorStatus;
+    }
+
+    /**
+     * Provides {@link TCACEFPolicyThresholdsProcessor} message
+     *
+     * @return threshold processor message
+     */
+    public String getThresholdCalculatorMessage() {
+        return thresholdCalculatorMessage;
+    }
+
+    /**
+     * Sets Threshold Calculator Processor Message
+     *
+     * @param thresholdCalculatorMessage threshold calculator message
+     */
+    public void setThresholdCalculatorMessage(String thresholdCalculatorMessage) {
+        this.thresholdCalculatorMessage = thresholdCalculatorMessage;
+    }
+
+    /**
+     * Provides generated alert message
+     *
+     * @return generated alert message
+     */
+    public String getAlertMessage() {
+        return alertMessage;
+    }
+
+    /**
+     * Sets alert message
+     *
+     * @param alertMessage alert message
+     */
+    public void setAlertMessage(String alertMessage) {
+        this.alertMessage = alertMessage;
+    }
+
+    /**
+     * Write entity to Table
+     *
+     * @param dataOutput data output
+     *
+     * @throws IOException io exception
+     */
+    @Override
+    public void write(DataOutput dataOutput) throws IOException {
+        WritableUtils.writeVLong(dataOutput, creationTS);
+        WritableUtils.writeVInt(dataOutput, instanceId);
+        WritableUtils.writeString(dataOutput, messageType);
+        WritableUtils.writeString(dataOutput, vesMessage);
+
+        WritableUtils.writeString(dataOutput, domain);
+        WritableUtils.writeString(dataOutput, functionalRole);
+
+        WritableUtils.writeString(dataOutput, thresholdPath);
+        WritableUtils.writeString(dataOutput, thresholdSeverity);
+        WritableUtils.writeString(dataOutput, thresholdDirection);
+        WritableUtils.writeVLong(dataOutput, thresholdValue);
+
+        WritableUtils.writeString(dataOutput, jsonProcessorStatus);
+        WritableUtils.writeString(dataOutput, jsonProcessorMessage);
+        WritableUtils.writeString(dataOutput, domainFilterStatus);
+        WritableUtils.writeString(dataOutput, domainFilterMessage);
+        WritableUtils.writeString(dataOutput, functionalRoleFilterStatus);
+        WritableUtils.writeString(dataOutput, functionalRoleFilterMessage);
+        WritableUtils.writeString(dataOutput, thresholdCalculatorStatus);
+        WritableUtils.writeString(dataOutput, thresholdCalculatorMessage);
+
+        WritableUtils.writeString(dataOutput, alertMessage);
+
+    }
+
+    /**
+     * Read entity from table
+     *
+     * @param dataInput data input
+     * @throws IOException io exception
+     */
+    @Override
+    public void readFields(DataInput dataInput) throws IOException {
+        creationTS = WritableUtils.readVLong(dataInput);
+        instanceId = WritableUtils.readVInt(dataInput);
+        messageType = WritableUtils.readString(dataInput);
+        vesMessage = WritableUtils.readString(dataInput);
+
+        domain = WritableUtils.readString(dataInput);
+        functionalRole = WritableUtils.readString(dataInput);
+
+        thresholdPath = WritableUtils.readString(dataInput);
+        thresholdSeverity = WritableUtils.readString(dataInput);
+        thresholdDirection = WritableUtils.readString(dataInput);
+        thresholdValue = WritableUtils.readVLong(dataInput);
+
+        jsonProcessorStatus = WritableUtils.readString(dataInput);
+        jsonProcessorMessage = WritableUtils.readString(dataInput);
+        domainFilterStatus = WritableUtils.readString(dataInput);
+        domainFilterMessage = WritableUtils.readString(dataInput);
+        functionalRoleFilterStatus = WritableUtils.readString(dataInput);
+        functionalRoleFilterMessage = WritableUtils.readString(dataInput);
+        thresholdCalculatorStatus = WritableUtils.readString(dataInput);
+        thresholdCalculatorMessage = WritableUtils.readString(dataInput);
+
+        alertMessage = WritableUtils.readString(dataInput);
+
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusPersister.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusPersister.java
new file mode 100644
index 0000000..0f4c539
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAMessageStatusPersister.java
@@ -0,0 +1,244 @@
+/*
+ * ===============================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.common.persistance.tca;
+
+import co.cask.cdap.api.data.schema.Schema;
+import co.cask.cdap.api.data.schema.UnsupportedTypeException;
+import co.cask.cdap.api.dataset.DatasetProperties;
+import co.cask.cdap.api.dataset.lib.IndexedTable;
+import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
+import co.cask.cdap.api.dataset.lib.ObjectMappedTableProperties;
+import com.google.common.base.Joiner;
+import org.apache.commons.lang3.StringEscapeUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.openecomp.dcae.apod.analytics.common.service.processor.MessageProcessor;
+import org.openecomp.dcae.apod.analytics.common.service.processor.ProcessorContext;
+import org.openecomp.dcae.apod.analytics.common.utils.PersistenceUtils;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.MetricsPerFunctionalRole;
+import org.openecomp.dcae.apod.analytics.model.domain.policy.tca.Threshold;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFJsonProcessor;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyDomainFilter;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyFunctionalRoleFilter;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFPolicyThresholdsProcessor;
+import org.openecomp.dcae.apod.analytics.tca.processor.TCACEFProcessorContext;
+import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.annotation.Nullable;
+
+import static org.openecomp.dcae.apod.analytics.common.utils.PersistenceUtils.TABLE_ROW_KEY_COLUMN_NAME;
+
+/**
+ *
+ *
+ * @author Rajiv Singla . Creation Date: 11/15/2016.
+ */
+public abstract class TCAMessageStatusPersister {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TCAMessageStatusPersister.class);
+
+    private TCAMessageStatusPersister() {
+
+    }
+
+    /**
+     * Saves Message Status in Table. Assumes no alert was generated
+     *
+     * @param processorContext processor Context
+     * @param instanceId Instance Id
+     * @param calculatorMessageType Calculation Message Type
+     * @param messageStatusTable Message Status Table
+     */
+    public static void persist(final TCACEFProcessorContext processorContext,
+                               final int instanceId,
+                               final TCACalculatorMessageType calculatorMessageType,
+                               final ObjectMappedTable<TCAMessageStatusEntity> messageStatusTable) {
+        persist(processorContext, instanceId, calculatorMessageType, messageStatusTable, null);
+    }
+
+    /**
+     * Saves Message Status in Table. Sets up alert message aslo
+     *
+     * @param processorContext processor Context
+     * @param instanceId Instance Id
+     * @param calculatorMessageType Calculation Message Type
+     * @param messageStatusTable Message Status Table
+     * @param alertMessage Alert message
+     */
+    public static void persist(final TCACEFProcessorContext processorContext,
+                               final int instanceId,
+                               final TCACalculatorMessageType calculatorMessageType,
+                               final ObjectMappedTable<TCAMessageStatusEntity> messageStatusTable,
+                               @Nullable final String alertMessage) {
+
+        final String rowKey = createKey(calculatorMessageType);
+
+        final Long currentTS = new Date().getTime();
+        final String vesMessage = StringEscapeUtils.unescapeJson(processorContext.getMessage());
+
+        // Find Functional Role and domain
+        final Pair<String, String> domainAndFunctionalRole = TCAUtils.getDomainAndFunctionalRole(processorContext);
+        final String domain = domainAndFunctionalRole.getLeft();
+        final String functionalRole = domainAndFunctionalRole.getRight();
+
+        final TCAMessageStatusEntity tcaMessageStatusEntity = new TCAMessageStatusEntity(currentTS,
+                instanceId, calculatorMessageType.name(), vesMessage, domain, functionalRole);
+
+        // add threshold violation fields
+        addViolatedThreshold(tcaMessageStatusEntity, processorContext);
+        // add processor status and messages
+        addMessageProcessorMessages(tcaMessageStatusEntity, processorContext);
+        // add Alert message
+        tcaMessageStatusEntity.setAlertMessage(
+                alertMessage == null ? null : StringEscapeUtils.unescapeJson(alertMessage)
+        );
+
+        messageStatusTable.write(rowKey, tcaMessageStatusEntity);
+
+        LOG.debug("Finished persisting VES Status Message with rowKey: {} in Message Status Table.", rowKey);
+
+    }
+
+
+    /**
+     * Create TCA VES Message Status Table Properties
+     *
+     * @param timeToLiveSeconds Message Status Table time to live in seconds
+     *
+     * @return Message Status table properties
+     */
+    public static DatasetProperties getDatasetProperties(final int timeToLiveSeconds) {
+
+        try {
+            return ObjectMappedTableProperties.builder()
+                    .setType(TCAMessageStatusEntity.class)
+                    .setRowKeyExploreName(TABLE_ROW_KEY_COLUMN_NAME)
+                    .setRowKeyExploreType(Schema.Type.STRING)
+                    .add(IndexedTable.PROPERTY_TTL, timeToLiveSeconds)
+                    .setDescription(CDAPComponentsConstants.TCA_FIXED_VES_MESSAGE_STATUS_DESCRIPTION_TABLE)
+                    .build();
+        } catch (UnsupportedTypeException e) {
+            final String errorMessage = "Unable to convert TCAMessageStatusEntity class to Schema";
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+
+    }
+
+
+    /**
+     * Adds Violated Threshold Parameter values to {@link TCAMessageStatusEntity}
+     *
+     * @param tcaMessageStatusEntity message entity that needs to be populated with threshold fields
+     * @param processorContext processor context
+     *
+     * @return entity with populated threshold field values if present
+     */
+    public static TCAMessageStatusEntity addViolatedThreshold(final TCAMessageStatusEntity tcaMessageStatusEntity,
+                                                               final TCACEFProcessorContext processorContext) {
+
+        final MetricsPerFunctionalRole metricsPerFunctionalRole = processorContext.getMetricsPerFunctionalRole();
+
+        if (metricsPerFunctionalRole != null
+                && metricsPerFunctionalRole.getThresholds() != null
+                && metricsPerFunctionalRole.getThresholds().get(0) != null) {
+
+            final Threshold threshold = metricsPerFunctionalRole.getThresholds().get(0);
+            tcaMessageStatusEntity.setThresholdPath(threshold.getFieldPath());
+            tcaMessageStatusEntity.setThresholdSeverity(threshold.getSeverity().name());
+            tcaMessageStatusEntity.setThresholdDirection(threshold.getDirection().name());
+            tcaMessageStatusEntity.setThresholdValue(threshold.getThresholdValue());
+        }
+
+        return tcaMessageStatusEntity;
+    }
+
+
+    /**
+     * Add TCA CEF Message Processor status information
+     *
+     * @param tcaMessageStatusEntity message entity that needs to be populated with message processor fields
+     * @param processorContext processor context
+     *
+     * @return entity with populated message process status information
+     */
+    public static TCAMessageStatusEntity addMessageProcessorMessages(
+            final TCAMessageStatusEntity tcaMessageStatusEntity, final TCACEFProcessorContext processorContext) {
+        final List<? super MessageProcessor<? extends ProcessorContext>> messageProcessors = processorContext
+                .getMessageProcessors();
+
+        if (messageProcessors != null && !messageProcessors.isEmpty()) {
+            for (Object messageProcessor : messageProcessors) {
+                final MessageProcessor<TCACEFProcessorContext> tcaMessageProcessor =
+                        (MessageProcessor<TCACEFProcessorContext>) messageProcessor;
+
+                final String processingState = tcaMessageProcessor.getProcessingState().name();
+                final String processingMessage = tcaMessageProcessor.getProcessingMessage().orNull();
+
+                if (messageProcessor.getClass().equals(TCACEFJsonProcessor.class)) {
+                    tcaMessageStatusEntity.setJsonProcessorStatus(processingState);
+                    tcaMessageStatusEntity.setJsonProcessorMessage(processingMessage);
+                }
+
+                if (messageProcessor.getClass().equals(TCACEFPolicyDomainFilter.class)) {
+                    tcaMessageStatusEntity.setDomainFilterStatus(processingState);
+                    tcaMessageStatusEntity.setDomainFilterMessage(processingMessage);
+                }
+
+                if (messageProcessor.getClass().equals(TCACEFPolicyFunctionalRoleFilter.class)) {
+                    tcaMessageStatusEntity.setFunctionalRoleFilterStatus(processingState);
+                    tcaMessageStatusEntity.setFunctionalRoleFilterMessage(processingMessage);
+                }
+
+                if (messageProcessor.getClass().equals(TCACEFPolicyThresholdsProcessor.class)) {
+                    tcaMessageStatusEntity.setThresholdCalculatorStatus(processingState);
+                    tcaMessageStatusEntity.setThresholdCalculatorMessage(processingMessage);
+                }
+
+            }
+        }
+        return tcaMessageStatusEntity;
+    }
+
+    /**
+     * Creates Row Key for TCA VES Message Status table
+     *
+     * Row Key = (Message Type + Decreasing Value)
+     *
+     * @param calculatorMessageType calculator message type
+     *
+     * @return row key string
+     */
+    public static String createKey(final TCACalculatorMessageType calculatorMessageType) {
+
+        final List<String> keyList = new LinkedList<>();
+        keyList.add(calculatorMessageType.name());
+        keyList.add(PersistenceUtils.getCurrentTimeReverseSubKey());
+        return Joiner.on(PersistenceUtils.ROW_KEY_DELIMITER).join(keyList);
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertEntity.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertEntity.java
new file mode 100644
index 0000000..7c333d8
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertEntity.java
@@ -0,0 +1,78 @@
+/*
+ * ===============================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.common.persistance.tca;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ *
+ * @author Rajiv Singla . Creation Date: 11/16/2016.
+ */
+public class TCAVESAlertEntity implements Writable, Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private long creationTS;
+    private String alertMessage;
+
+    public TCAVESAlertEntity() {
+        // no argument constructor required for json serialization / deserialization
+    }
+
+    public TCAVESAlertEntity(long creationTS, String alertMessage) {
+        this.creationTS = creationTS;
+        this.alertMessage = alertMessage;
+    }
+
+    public long getCreationTS() {
+        return creationTS;
+    }
+
+    public void setCreationTS(long creationTS) {
+        this.creationTS = creationTS;
+    }
+
+    public String getAlertMessage() {
+        return alertMessage;
+    }
+
+    public void setAlertMessage(String alertMessage) {
+        this.alertMessage = alertMessage;
+    }
+
+    @Override
+    public void write(DataOutput dataOutput) throws IOException {
+        WritableUtils.writeVLong(dataOutput, creationTS);
+        WritableUtils.writeString(dataOutput, alertMessage);
+    }
+
+    @Override
+    public void readFields(DataInput dataInput) throws IOException {
+        creationTS = WritableUtils.readVLong(dataInput);
+        alertMessage = WritableUtils.readString(dataInput);
+    }
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertsPersister.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertsPersister.java
new file mode 100644
index 0000000..18293c7
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/persistance/tca/TCAVESAlertsPersister.java
@@ -0,0 +1,102 @@
+/*
+ * ===============================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.common.persistance.tca;
+
+import co.cask.cdap.api.data.schema.Schema;
+import co.cask.cdap.api.data.schema.UnsupportedTypeException;
+import co.cask.cdap.api.dataset.DatasetProperties;
+import co.cask.cdap.api.dataset.lib.IndexedTable;
+import co.cask.cdap.api.dataset.lib.ObjectMappedTable;
+import co.cask.cdap.api.dataset.lib.ObjectMappedTableProperties;
+import org.apache.commons.lang3.StringEscapeUtils;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+
+import static org.openecomp.dcae.apod.analytics.common.utils.PersistenceUtils.TABLE_ROW_KEY_COLUMN_NAME;
+
+/**
+ *
+ * @author Rajiv Singla . Creation Date: 11/16/2016.
+ */
+public abstract class TCAVESAlertsPersister {
+
+    private static final Logger LOG = LoggerFactory.getLogger(TCAVESAlertsPersister.class);
+
+    private TCAVESAlertsPersister() {
+
+    }
+
+    /**
+     * Persists Alert Message to Alerts Table
+     *
+     * @param alertMessage alert Message
+     * @param tcaVESAlertTable alert Table Name
+     */
+    public static void persist(final String alertMessage, final ObjectMappedTable<TCAVESAlertEntity> tcaVESAlertTable) {
+        final Date currentDate = new Date();
+        final TCAVESAlertEntity alertEntity = new TCAVESAlertEntity(currentDate.getTime(),
+                StringEscapeUtils.unescapeJson(alertMessage));
+        // row key is same as current timestamp
+        final String rowKey = createRowKey(currentDate);
+        tcaVESAlertTable.write(rowKey, alertEntity);
+
+        LOG.debug("Finished persisting VES Alert message ID: {} in VES Alerts table.", rowKey);
+    }
+
+
+    /**
+     * Creates {@link DatasetProperties} for Alerts Table
+     *
+     * @param timeToLiveSeconds alerts table Time to Live
+     * @return Alerts table properties
+     */
+    public static DatasetProperties getDatasetProperties(final int timeToLiveSeconds) {
+        try {
+            return ObjectMappedTableProperties.builder()
+                    .setType(TCAVESAlertEntity.class)
+                    .setRowKeyExploreName(TABLE_ROW_KEY_COLUMN_NAME)
+                    .setRowKeyExploreType(Schema.Type.STRING)
+                    .add(IndexedTable.PROPERTY_TTL, timeToLiveSeconds)
+                    .setDescription(CDAPComponentsConstants.TCA_DEFAULT_VES_ALERTS_DESCRIPTION_TABLE)
+                    .build();
+        } catch (UnsupportedTypeException e) {
+            final String errorMessage = "Unable to convert TCAVESAlertEntity class to Schema";
+            throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+        }
+
+    }
+
+    /**
+     * Creates Row Key for Alerts Table
+     *
+     * @param date current Date
+     *
+     * @return row key
+     */
+    public static String createRowKey(final Date date) {
+       return String.format("%025d", date.getTime());
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppConfig.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppConfig.java
new file mode 100644
index 0000000..c3254d0
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppConfig.java
@@ -0,0 +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.cdap.common.settings;
+
+/**
+ *<p>
+ *     Minimum Contract for all CDAP App Configs
+ *</p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/2/2016.
+ */
+public interface CDAPAppConfig extends CDAPAppSettings {
+
+    /**
+     *  CDAP Application Name
+     *
+     * @return cdap app name
+     */
+    String getAppName();
+
+
+    /**
+     *  CDAP Application Description
+     *
+     * @return cdap app description
+     */
+    String getAppDescription();
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppPreferences.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppPreferences.java
new file mode 100644
index 0000000..26433cd
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppPreferences.java
@@ -0,0 +1,31 @@
+/*
+ * ===============================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.common.settings;
+
+/**
+ * <p>
+ *     Marker Interface for all CDAP Preferences
+ * </p>
+ *
+ * @author Rajiv Singla . Creation Date: 11/2/2016.
+ */
+public interface CDAPAppPreferences extends CDAPAppSettings {
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppSettings.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppSettings.java
new file mode 100644
index 0000000..29ede41
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPAppSettings.java
@@ -0,0 +1,35 @@
+/*
+ * ===============================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.common.settings;
+
+import java.io.Serializable;
+
+/**
+ * <p>
+ *      A marker interface for all CDAP Related App Settings.
+ *      App Settings can either be preferences, cdap app settings etc.
+ * </p>
+ *
+ *
+ * @author Rajiv Singla . Creation Date: 11/2/2016.
+ */
+public interface CDAPAppSettings extends Serializable {
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBaseAppConfig.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBaseAppConfig.java
new file mode 100644
index 0000000..f062825
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBaseAppConfig.java
@@ -0,0 +1,64 @@
+/*
+ * ===============================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.common.settings;
+
+import co.cask.cdap.api.Config;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+
+/**
+ * Base class for all DACE Analytics Application Configurations
+ *
+ * @author Rajiv Singla . Creation Date: 10/4/2016.
+ */
+public abstract class CDAPBaseAppConfig extends Config implements CDAPAppConfig {
+
+
+    /**
+     * DCAE Analytics App Name
+     */
+    protected String appName = CDAPComponentsConstants.COMMON_DEFAULT_DCAE_CDAP_NAME_APP;
+
+    /**
+     * DCAE Analytics App Description
+     */
+    protected String appDescription = CDAPComponentsConstants.COMMON_DEFAULT_DCAE_CDAP_DESCRIPTION_APP;
+
+
+    /**
+     * Returns DCAE Analytics CDAP Application name
+     *
+     * @return CDAP application name
+     */
+    @Override
+    public String getAppName() {
+        return appName;
+    }
+
+    /**
+     * Returns DCAE Analytics CDAP Application descrption
+     *
+     * @return CDAP application description
+     */
+    @Override
+    public String getAppDescription() {
+        return appDescription;
+    }
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBasePluginConfig.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBasePluginConfig.java
new file mode 100644
index 0000000..a293175
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPBasePluginConfig.java
@@ -0,0 +1,53 @@
+/*
+ * ===============================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.common.settings;
+
+import co.cask.cdap.api.annotation.Description;
+import co.cask.cdap.api.annotation.Macro;
+import co.cask.cdap.api.annotation.Name;
+import co.cask.cdap.api.plugin.PluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPPluginConstants;
+
+/**
+ * <p>
+ *      Base class for all DCAE Analytics CDAP Plugin config
+ * </p>
+ * <p>
+ * @author Rajiv Singla . Creation Date: 1/17/2017.
+ */
+public abstract class CDAPBasePluginConfig extends PluginConfig implements CDAPPluginSettings {
+
+    @Name(CDAPPluginConstants.Reference.REFERENCE_NAME)
+    @Description(CDAPPluginConstants.Reference.REFERENCE_NAME_DESCRIPTION)
+    @Macro
+    protected String referenceName;
+
+    /**
+     * Provides Reference Name that can be used to trace lineage or meta data information inside CDAP container
+     *
+     * @return reference name
+     */
+    public String getReferenceName() {
+        return referenceName;
+    }
+
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPPluginSettings.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPPluginSettings.java
new file mode 100644
index 0000000..dcb0218
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/settings/CDAPPluginSettings.java
@@ -0,0 +1,31 @@
+/*
+ * ===============================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.common.settings;
+
+/**
+ * <p>
+ *      A marker interface for all CDAP Plugin related Settings.
+ * </p>
+ *
+ * @author Rajiv Singla . Creation Date: 1/17/2017.
+ */
+public interface CDAPPluginSettings extends CDAPAppSettings {
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/DMaaPMRUtils.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/DMaaPMRUtils.java
new file mode 100644
index 0000000..37acf9c
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/DMaaPMRUtils.java
@@ -0,0 +1,144 @@
+/*
+ * ===============================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.common.utils;
+
+import co.cask.cdap.api.metrics.Metrics;
+import com.google.common.base.Optional;
+import com.google.common.base.Stopwatch;
+import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
+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.response.DMaaPMRSubscriberResponse;
+import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * Utility common methods for DMaaP MR functionality
+ *
+ * @author Rajiv Singla . Creation Date: 2/6/2017.
+ */
+public abstract class DMaaPMRUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(DMaaPMRUtils.class);
+
+    private DMaaPMRUtils() {
+        // private constructor
+    }
+
+
+    /**
+     * Returns messages fetched from DMaaP MR Subscriber.
+     *
+     * @param subscriber DMaaP MR Subscriber instance
+     * @param metrics CDAP metrics
+     *
+     * @return messages fetched from DMaaP MR topic
+     */
+    public static Optional<List<String>> getSubscriberMessages(final DMaaPMRSubscriber subscriber,
+                                                               final Metrics metrics) {
+
+        final Optional<DMaaPMRSubscriberResponse> subscriberResponseOptional =
+                getSubscriberResponse(subscriber, metrics);
+
+        // If response is not present, unable to proceed
+        if (!subscriberResponseOptional.isPresent()) {
+            return Optional.absent();
+        }
+
+        final DMaaPMRSubscriberResponse subscriberResponse = subscriberResponseOptional.get();
+
+        // If response code return by the subscriber call is not successful, unable to do proceed
+        if (!HTTPUtils.isSuccessfulResponseCode(subscriberResponse.getResponseCode())) {
+            LOG.error("Subscriber was unable to fetch messages properly.Subscriber Response Code: {} " +
+                    "Unable to proceed further....", subscriberResponse.getResponseCode());
+            metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_UNSUCCESSFUL_RESPONSES_METRIC, 1);
+            return Optional.absent();
+        }
+
+        LOG.debug("Subscriber HTTP Response Status Code match successful:  {}", subscriberResponse,
+                HTTPUtils.HTTP_SUCCESS_STATUS_CODE);
+
+        final List<String> actualMessages = subscriberResponse.getFetchedMessages();
+
+        // If there are no message returned during from Subscriber, nothing to write to CDAP Stream
+        if (actualMessages.isEmpty()) {
+            LOG.debug("Subscriber Response has no messages. Nothing to write....");
+            metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_RESPONSES_WITH_NO_MESSAGES_METRIC, 1);
+            return Optional.absent();
+        }
+
+        LOG.debug("DMaaP MR Subscriber found new messages in DMaaP Topic. Message count: {}", actualMessages.size());
+        metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_TOTAL_MESSAGES_PROCESSED_METRIC, actualMessages.size());
+
+        return Optional.of(actualMessages);
+
+    }
+
+
+    /**
+     * Get Subscriber response and records time taken to fetch messages. Returns Optional.None if Subscriber response
+     * is null or response status code is not present
+     *
+     * @param subscriber - DMaaP Subscriber
+     * @param metrics - CDAP Metrics collector
+     *
+     * @return - Optional of Subscriber Response
+     */
+    public static Optional<DMaaPMRSubscriberResponse> getSubscriberResponse(final DMaaPMRSubscriber subscriber,
+                                                                            final Metrics metrics) {
+
+        // Record all response count from subscriber
+        metrics.count(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_ALL_RESPONSES_COUNT_METRIC, 1);
+
+        // Check how long it took for subscriber to respond
+        final Stopwatch stopwatch = new Stopwatch();
+        stopwatch.start();
+
+        // Fetch messages from DMaaP MR Topic
+        DMaaPMRSubscriberResponse subscriberResponse = null;
+        try {
+            subscriberResponse = subscriber.fetchMessages();
+        } catch (DCAEAnalyticsRuntimeException e) {
+            LOG.error("Error while fetching messages for DMaaP MR Topic: {}", e);
+        }
+
+        stopwatch.stop();
+        final long subscriberResponseTimeMS = stopwatch.elapsedMillis();
+
+        // If response is null is null or response code is null, unable to proceed nothing to do
+        if (subscriberResponse == null || subscriberResponse.getResponseCode() == null) {
+            LOG.error("Subscriber Response is null or subscriber Response code is null. Unable to proceed further...");
+            return Optional.absent();
+        }
+
+        LOG.debug("Subscriber Response:{}, Subscriber HTTP Response Status Code {}, Subscriber Response Time(ms): {}",
+                subscriberResponse, subscriberResponse.getResponseCode(), subscriberResponseTimeMS);
+
+        // Record subscriber response time
+        metrics.gauge(CDAPMetricsConstants.DMAAP_MR_SUBSCRIBER_RESPONSE_TIME_MS_METRIC, subscriberResponseTimeMS);
+
+        return Optional.of(subscriberResponse);
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/ValidationUtils.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/ValidationUtils.java
new file mode 100644
index 0000000..b987caf
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/utils/ValidationUtils.java
@@ -0,0 +1,107 @@
+/*
+ * ===============================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.common.utils;
+
+
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.common.settings.CDAPAppSettings;
+import org.openecomp.dcae.apod.analytics.cdap.common.validation.CDAPAppSettingsValidator;
+import org.openecomp.dcae.apod.analytics.common.validation.ValidationResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Utility methods to validate null checks, empty string etc
+ *
+ * @author Rajiv Singla . Creation Date: 10/24/2016.
+ */
+public abstract class ValidationUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ValidationUtils.class);
+
+    private ValidationUtils() {
+
+    }
+
+    /**
+     * Checks if String is empty. For null string true is returned
+     *
+     * @param stringValue string value
+     * @return returns true is string is empty or null
+     */
+    public static boolean isEmpty(@Nullable final String stringValue) {
+        return stringValue == null || stringValue.isEmpty() || stringValue.trim().isEmpty();
+    }
+
+
+    /**
+     * Checks if String value is present. A null, empty, or blank values of string
+     * are considered not present.
+     *
+     * @param stringValue string value to check if it is present or not
+     *
+     * @return true if string value is not null, empty or blank
+     */
+    public static boolean isPresent(@Nullable final String stringValue) {
+        return !isEmpty(stringValue);
+    }
+
+
+    /**
+     * Provides common functionality to Validates CDAP App Settings. Throws Runtime exception if validation fails
+     *
+     * @param appSettings app Settings e.g. App Config, App Preferences etc
+     * @param appSettingsValidator app Settings validator
+     *
+     * @param <T> Settings type e.g. AppConfig or AppPreferences
+     * @param <R> Validation Response type
+     * @param <V> Validator Type
+     */
+    public static <T extends CDAPAppSettings, R extends ValidationResponse<T>,
+            V extends CDAPAppSettingsValidator<T, R>> void validateSettings(@Nonnull final T appSettings,
+                                                                            @Nonnull final V appSettingsValidator) {
+        checkNotNull(appSettings, "App Settings must not be null");
+        checkNotNull(appSettingsValidator, "App Settings validator must not be null");
+
+        final String appSettingsClassName = appSettings.getClass().getSimpleName();
+        final String appSettingsClassValidator = appSettingsValidator.getClass().getSimpleName();
+
+        LOG.debug("Validating App Settings for: {}, with App Settings Validator: {} ",
+                appSettingsClassName, appSettingsClassValidator);
+
+        final R validationResponse = appSettingsValidator.validateAppSettings(appSettings);
+
+        // If setting validation fails throw an exception
+        if (validationResponse.hasErrors()) {
+            throw new CDAPSettingsException(
+                    validationResponse.getAllErrorMessage(), LOG, new IllegalArgumentException());
+        }
+
+        LOG.debug("App Settings Validation Successful for app Settings: {} with validator: {}", appSettingsClassName,
+                appSettingsClassValidator);
+    }
+
+}
diff --git a/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/validation/CDAPAppSettingsValidator.java b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/validation/CDAPAppSettingsValidator.java
new file mode 100644
index 0000000..48689f7
--- /dev/null
+++ b/dcae-analytics-cdap-common/src/main/java/org/openecomp/dcae/apod/analytics/cdap/common/validation/CDAPAppSettingsValidator.java
@@ -0,0 +1,52 @@
+/*
+ * ===============================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.common.validation;
+
+import org.openecomp.dcae.apod.analytics.cdap.common.settings.CDAPAppSettings;
+import org.openecomp.dcae.apod.analytics.common.validation.DCAEValidator;
+import org.openecomp.dcae.apod.analytics.common.validation.ValidationResponse;
+
+/**
+ * <p>
+ *     Validates CDAP Application Settings (AppConfig, Preferences etc)
+ * <p>
+ *
+ * @param <T> {@link CDAPAppSettings} DCAE Analytics App Settings (e.g. AppConfig, Preferences)
+ * @param <R> {@link ValidationResponse} Validator response implementations
+ *
+ * @author Rajiv Singla . Creation Date: 11/2/2016.
+ */
+public interface CDAPAppSettingsValidator<T extends CDAPAppSettings, R extends ValidationResponse<T>>
+        extends DCAEValidator {
+
+    /**
+     * Validates DCAE Analytics App Settings and return Validation response which can be
+     * checked for any app setting issues
+     *
+     * @param appSettings DCAE CDAP Application Settings (e.g. AppConfig, Preferences etc.)
+     * @return validation response
+     */
+    R validateAppSettings(T appSettings);
+
+
+}
+
+