-/*\r
- * ===============================LICENSE_START======================================\r
- * dcae-analytics\r
- * ================================================================================\r
- * Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
- * ================================================================================\r
- * Licensed under the Apache License, Version 2.0 (the "License");\r
- * you may not use this file except in compliance with the License.\r
- * You may obtain a copy of the License at\r
- *\r
- * http://www.apache.org/licenses/LICENSE-2.0\r
- *\r
- * Unless required by applicable law or agreed to in writing, software\r
- * distributed under the License is distributed on an "AS IS" BASIS,\r
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * See the License for the specific language governing permissions and\r
- * limitations under the License.\r
- * ============================LICENSE_END===========================================\r
- */\r
-\r
-package org.openecomp.dcae.apod.analytics.cdap.tca.worker;\r
-\r
-import co.cask.cdap.api.metrics.Metrics;\r
-import co.cask.cdap.api.worker.WorkerContext;\r
-import com.google.common.base.Optional;\r
-import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;\r
-import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;\r
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;\r
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;\r
-import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;\r
-import org.quartz.DisallowConcurrentExecution;\r
-import org.quartz.Job;\r
-import org.quartz.JobDataMap;\r
-import org.quartz.JobExecutionContext;\r
-import org.quartz.JobExecutionException;\r
-import org.quartz.PersistJobDataAfterExecution;\r
-import org.slf4j.Logger;\r
-import org.slf4j.LoggerFactory;\r
-\r
-import java.io.IOException;\r
-import java.util.List;\r
-\r
-import static java.lang.String.format;\r
-\r
-/**\r
- * Quartz Job which polls DMaaP MR VES Collector Topic for messages and writes them to\r
- * a given CDAP Stream\r
- *\r
- * @author Rajiv Singla . Creation Date: 10/24/2016.\r
- */\r
-@DisallowConcurrentExecution\r
-@PersistJobDataAfterExecution\r
-public class TCADMaaPMRSubscriberJob implements Job {\r
-\r
- private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMRSubscriberJob.class);\r
-\r
- @Override\r
- public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {\r
-\r
- LOG.debug("Starting DMaaP MR Topic Subscriber fetch Job. Next firing time will be: {}",\r
- jobExecutionContext.getNextFireTime());\r
-\r
- // Get Job Data Map\r
- final JobDataMap jobDataMap = jobExecutionContext.getMergedJobDataMap();\r
-\r
- // Fetch all Job Params from Job Data Map\r
- final String cdapStreamName = jobDataMap.getString(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME);\r
- final WorkerContext workerContext =\r
- (WorkerContext) jobDataMap.get(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME);\r
- final DMaaPMRSubscriber subscriber =\r
- (DMaaPMRSubscriber) jobDataMap.get(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME);\r
- final Metrics metrics = (Metrics) jobDataMap.get(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME);\r
-\r
- final Optional<List<String>> subscriberMessagesOptional =\r
- DMaaPMRUtils.getSubscriberMessages(subscriber, metrics);\r
-\r
- // Write message to CDAP Stream using Stream Writer\r
- if (subscriberMessagesOptional.isPresent()) {\r
- writeMessageToCDAPStream(subscriberMessagesOptional.get(), cdapStreamName, workerContext, metrics);\r
- }\r
- }\r
-\r
-\r
- /**\r
- * Writes given messages to CDAP Stream\r
- *\r
- * @param actualMessages List of messages that need to written to cdap stream\r
- * @param cdapStreamName cdap stream name\r
- * @param workerContext cdap worker context\r
- * @param metrics cdap metrics\r
- */\r
- private void writeMessageToCDAPStream(final List<String> actualMessages, final String cdapStreamName,\r
- final WorkerContext workerContext, final Metrics metrics) {\r
- LOG.debug("Writing message to CDAP Stream: {}, Message Count: {}", cdapStreamName, actualMessages.size());\r
- try {\r
-\r
- for (String message : actualMessages) {\r
- workerContext.write(cdapStreamName, message);\r
- }\r
-\r
- } catch (IOException e) {\r
- metrics.count(CDAPMetricsConstants.TCA_SUBSCRIBER_FAILURE_TO_WRITE_TO_STREAM_METRIC, 1);\r
- final String errorMessage =\r
- format("Error while DMaaP message router subscriber attempting to write to CDAP Stream: %s, " +\r
- "Exception: %s", cdapStreamName, e);\r
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
- }\r
-\r
- LOG.debug("DMaaP MR Subscriber successfully finished writing messages to CDAP Stream: {}, Message count: {}",\r
- cdapStreamName, actualMessages.size());\r
-\r
- }\r
-\r
-}\r
+/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.onap.dcae.apod.analytics.cdap.tca.worker;
+
+import co.cask.cdap.api.metrics.Metrics;
+import co.cask.cdap.api.worker.WorkerContext;
+import com.google.common.base.Optional;
+import org.onap.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
+import org.onap.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.PersistJobDataAfterExecution;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+import static java.lang.String.format;
+
+/**
+ * Quartz Job which polls DMaaP MR VES Collector Topic for messages and writes them to
+ * a given CDAP Stream
+ *
+ * @author Rajiv Singla . Creation Date: 10/24/2016.
+ */
+@DisallowConcurrentExecution
+@PersistJobDataAfterExecution
+public class TCADMaaPMRSubscriberJob implements Job {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMRSubscriberJob.class);
+
+ @Override
+ public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+
+ LOG.debug("Starting DMaaP MR Topic Subscriber fetch Job. Next firing time will be: {}",
+ jobExecutionContext.getNextFireTime());
+
+ // Get Job Data Map
+ final JobDataMap jobDataMap = jobExecutionContext.getMergedJobDataMap();
+
+ // Fetch all Job Params from Job Data Map
+ final String cdapStreamName = jobDataMap.getString(AnalyticsConstants.CDAP_STREAM_VARIABLE_NAME);
+ final WorkerContext workerContext =
+ (WorkerContext) jobDataMap.get(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME);
+ final DMaaPMRSubscriber subscriber =
+ (DMaaPMRSubscriber) jobDataMap.get(AnalyticsConstants.DMAAP_SUBSCRIBER_VARIABLE_NAME);
+ final Metrics metrics = (Metrics) jobDataMap.get(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME);
+
+ final Optional<List<String>> subscriberMessagesOptional =
+ DMaaPMRUtils.getSubscriberMessages(subscriber, metrics);
+
+ // Write message to CDAP Stream using Stream Writer
+ if (subscriberMessagesOptional.isPresent()) {
+ writeMessageToCDAPStream(subscriberMessagesOptional.get(), cdapStreamName, workerContext, metrics);
+ }
+ }
+
+
+ /**
+ * Writes given messages to CDAP Stream
+ *
+ * @param actualMessages List of messages that need to written to cdap stream
+ * @param cdapStreamName cdap stream name
+ * @param workerContext cdap worker context
+ * @param metrics cdap metrics
+ */
+ private void writeMessageToCDAPStream(final List<String> actualMessages, final String cdapStreamName,
+ final WorkerContext workerContext, final Metrics metrics) {
+ LOG.debug("Writing message to CDAP Stream: {}, Message Count: {}", cdapStreamName, actualMessages.size());
+ try {
+
+ for (String message : actualMessages) {
+ workerContext.write(cdapStreamName, message);
+ }
+
+ } catch (IOException e) {
+ metrics.count(CDAPMetricsConstants.TCA_SUBSCRIBER_FAILURE_TO_WRITE_TO_STREAM_METRIC, 1);
+ final String errorMessage =
+ format("Error while DMaaP message router subscriber attempting to write to CDAP Stream: %s, " +
+ "Exception: %s", cdapStreamName, e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+
+ LOG.debug("DMaaP MR Subscriber successfully finished writing messages to CDAP Stream: {}, Message count: {}",
+ cdapStreamName, actualMessages.size());
+
+ }
+
+}
Code Review / dcaegen2 / analytics / tca.git / blobdiff