-/*
- * ===============================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.tca.worker;
-
-import co.cask.cdap.api.metrics.Metrics;
-import co.cask.cdap.api.worker.WorkerContext;
-import com.google.common.base.Optional;
-import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants;
-import org.openecomp.dcae.apod.analytics.cdap.common.utils.DMaaPMRUtils;
-import org.openecomp.dcae.apod.analytics.common.AnalyticsConstants;
-import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
-import org.openecomp.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());
-
- }
-
-}
+/*\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
Code Review / dcaegen2 / analytics / tca.git / blobdiff