-/*\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.annotation.Property;\r
-import co.cask.cdap.api.metrics.Metrics;\r
-import co.cask.cdap.api.worker.WorkerContext;\r
-import org.openecomp.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;\r
-import org.openecomp.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;\r
-import org.openecomp.dcae.apod.analytics.cdap.tca.utils.AppPreferencesToPublisherConfigMapper;\r
-import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;\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.DMaaPMRFactory;\r
-import org.openecomp.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;\r
-import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;\r
-import org.openecomp.dcae.apod.analytics.tca.utils.TCAUtils;\r
-import org.quartz.JobDataMap;\r
-import org.quartz.SchedulerException;\r
-import org.quartz.impl.StdSchedulerFactory;\r
-import org.slf4j.Logger;\r
-import org.slf4j.LoggerFactory;\r
-\r
-import java.util.concurrent.atomic.AtomicBoolean;\r
-\r
-import static java.lang.String.format;\r
-\r
-/**\r
- * TCA DMaaP Publisher will monitor alerts table at regular intervals and publish any alerts to DMaaP MR Publishing\r
- * Topic\r
- * <p>\r
- * @author Rajiv Singla . Creation Date: 11/16/2016.\r
- */\r
-public class TCADMaaPPublisherWorker extends BaseTCADMaaPMRWorker {\r
-\r
- private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPPublisherWorker.class);\r
-\r
- private DMaaPMRPublisher publisher;\r
- private Metrics metrics;\r
- @Property\r
- private final String tcaVESAlertsTableName;\r
-\r
- public TCADMaaPPublisherWorker(final String tcaVESAlertsTableName) {\r
- this.tcaVESAlertsTableName = tcaVESAlertsTableName;\r
- }\r
-\r
- @Override\r
- public void configure() {\r
- setName(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);\r
- setDescription(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_DESCRIPTION_WORKER);\r
- LOG.debug("Configuring TCA MR DMaaP Publisher worker with name: {}",\r
- CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);\r
- }\r
-\r
-\r
- @Override\r
- public void initialize(WorkerContext context) throws Exception {\r
- super.initialize(context);\r
-\r
- // Parse runtime arguments\r
- final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context);\r
-\r
- LOG.info("Initializing TCA MR DMaaP Publisher worker with preferences: {}", tcaAppPreferences);\r
-\r
- // Map TCA App Preferences to DMaaP MR Publisher Config\r
- final DMaaPMRPublisherConfig publisherConfig = AppPreferencesToPublisherConfigMapper.map(tcaAppPreferences);\r
-\r
- LOG.info("TCA DMaaP MR Publisher worker will be polling TCA Alerts Table Name: {}", tcaVESAlertsTableName);\r
-\r
- // Create an instance of DMaaP MR Publisher\r
- LOG.debug("Creating an instance of DMaaP Publisher");\r
- publisher = DMaaPMRFactory.create().createPublisher(publisherConfig);\r
-\r
- // initialize a new Quartz scheduler\r
- initializeScheduler(tcaAppPreferences, new StdSchedulerFactory());\r
-\r
- // initialize scheduler state\r
- isSchedulerShutdown = new AtomicBoolean(true);\r
- }\r
-\r
-\r
- /**\r
- * Stop DMaaP Publisher\r
- */\r
- @Override\r
- public void stop() {\r
- // Close Publisher - which will flush any batch messages if present in batch queue\r
- if (publisher != null) {\r
- try {\r
- publisher.close();\r
- } catch (Exception e) {\r
- final String errorMessage = format("Error while shutting down DMaaP MR Publisher: %s", e);\r
- throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);\r
- }\r
- }\r
- // Shut down scheduler\r
- super.stop();\r
- }\r
-\r
-\r
- /**\r
- * Initializes a scheduler instance for DMaaP MR Publisher Job\r
- *\r
- * @throws SchedulerException SchedulerException\r
- */\r
- private void initializeScheduler(final TCAAppPreferences tcaAnalyticsAppConfig,\r
- final StdSchedulerFactory stdSchedulerFactory) throws SchedulerException {\r
-\r
- // Get Publisher polling interval\r
- final Integer publisherPollingInterval = tcaAnalyticsAppConfig.getPublisherPollingInterval();\r
-\r
- // Publisher Quartz Properties file\r
- final String quartzPublisherPropertiesFileName = AnalyticsConstants.TCA_QUARTZ_PUBLISHER_PROPERTIES_FILE_NAME;\r
-\r
- // Create a new JobDataMap containing information required by TCA DMaaP Publisher Job\r
- final JobDataMap jobDataMap = new JobDataMap();\r
- jobDataMap.put(AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME, tcaVESAlertsTableName);\r
- jobDataMap.put(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME, getContext());\r
- jobDataMap.put(AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME, publisher);\r
- jobDataMap.put(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME, metrics);\r
-\r
- // Create new publisher scheduler\r
- scheduler = TCAUtils.createQuartzScheduler(publisherPollingInterval, stdSchedulerFactory,\r
- quartzPublisherPropertiesFileName, jobDataMap, TCADMaaPMRPublisherJob.class,\r
- AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_JOB_NAME,\r
- AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_TRIGGER_NAME);\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.annotation.Property;
+import co.cask.cdap.api.metrics.Metrics;
+import co.cask.cdap.api.worker.WorkerContext;
+import org.onap.dcae.apod.analytics.cdap.common.CDAPComponentsConstants;
+import org.onap.dcae.apod.analytics.cdap.tca.settings.TCAAppPreferences;
+import org.onap.dcae.apod.analytics.cdap.tca.utils.AppPreferencesToPublisherConfigMapper;
+import org.onap.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils;
+import org.onap.dcae.apod.analytics.common.AnalyticsConstants;
+import org.onap.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.onap.dcae.apod.analytics.dmaap.DMaaPMRFactory;
+import org.onap.dcae.apod.analytics.dmaap.domain.config.DMaaPMRPublisherConfig;
+import org.onap.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher;
+import org.onap.dcae.apod.analytics.tca.utils.TCAUtils;
+import org.quartz.JobDataMap;
+import org.quartz.SchedulerException;
+import org.quartz.impl.StdSchedulerFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static java.lang.String.format;
+
+/**
+ * TCA DMaaP Publisher will monitor alerts table at regular intervals and publish any alerts to DMaaP MR Publishing
+ * Topic
+ * <p>
+ * @author Rajiv Singla . Creation Date: 11/16/2016.
+ */
+public class TCADMaaPPublisherWorker extends BaseTCADMaaPMRWorker {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPPublisherWorker.class);
+
+ private DMaaPMRPublisher publisher;
+ private Metrics metrics;
+ @Property
+ private final String tcaVESAlertsTableName;
+
+ public TCADMaaPPublisherWorker(final String tcaVESAlertsTableName) {
+ this.tcaVESAlertsTableName = tcaVESAlertsTableName;
+ }
+
+ @Override
+ public void configure() {
+ setName(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);
+ setDescription(CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_DESCRIPTION_WORKER);
+ LOG.debug("Configuring TCA MR DMaaP Publisher worker with name: {}",
+ CDAPComponentsConstants.TCA_FIXED_DMAAP_PUBLISHER_WORKER);
+ }
+
+
+ @Override
+ public void initialize(WorkerContext context) throws Exception {
+ super.initialize(context);
+
+ // Parse runtime arguments
+ final TCAAppPreferences tcaAppPreferences = CDAPTCAUtils.getValidatedTCAAppPreferences(context);
+
+ LOG.info("Initializing TCA MR DMaaP Publisher worker with preferences: {}", tcaAppPreferences);
+
+ // Map TCA App Preferences to DMaaP MR Publisher Config
+ final DMaaPMRPublisherConfig publisherConfig = AppPreferencesToPublisherConfigMapper.map(tcaAppPreferences);
+
+ LOG.info("TCA DMaaP MR Publisher worker will be polling TCA Alerts Table Name: {}", tcaVESAlertsTableName);
+
+ // Create an instance of DMaaP MR Publisher
+ LOG.debug("Creating an instance of DMaaP Publisher");
+ publisher = DMaaPMRFactory.create().createPublisher(publisherConfig);
+
+ // initialize a new Quartz scheduler
+ initializeScheduler(tcaAppPreferences, new StdSchedulerFactory());
+
+ // initialize scheduler state
+ isSchedulerShutdown = new AtomicBoolean(true);
+ }
+
+
+ /**
+ * Stop DMaaP Publisher
+ */
+ @Override
+ public void stop() {
+ // Close Publisher - which will flush any batch messages if present in batch queue
+ if (publisher != null) {
+ try {
+ publisher.close();
+ } catch (Exception e) {
+ final String errorMessage = format("Error while shutting down DMaaP MR Publisher: %s", e);
+ throw new DCAEAnalyticsRuntimeException(errorMessage, LOG, e);
+ }
+ }
+ // Shut down scheduler
+ super.stop();
+ }
+
+
+ /**
+ * Initializes a scheduler instance for DMaaP MR Publisher Job
+ *
+ * @throws SchedulerException SchedulerException
+ */
+ private void initializeScheduler(final TCAAppPreferences tcaAnalyticsAppConfig,
+ final StdSchedulerFactory stdSchedulerFactory) throws SchedulerException {
+
+ // Get Publisher polling interval
+ final Integer publisherPollingInterval = tcaAnalyticsAppConfig.getPublisherPollingInterval();
+
+ // Publisher Quartz Properties file
+ final String quartzPublisherPropertiesFileName = AnalyticsConstants.TCA_QUARTZ_PUBLISHER_PROPERTIES_FILE_NAME;
+
+ // Create a new JobDataMap containing information required by TCA DMaaP Publisher Job
+ final JobDataMap jobDataMap = new JobDataMap();
+ jobDataMap.put(AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME, tcaVESAlertsTableName);
+ jobDataMap.put(AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME, getContext());
+ jobDataMap.put(AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME, publisher);
+ jobDataMap.put(AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME, metrics);
+
+ // Create new publisher scheduler
+ scheduler = TCAUtils.createQuartzScheduler(publisherPollingInterval, stdSchedulerFactory,
+ quartzPublisherPropertiesFileName, jobDataMap, TCADMaaPMRPublisherJob.class,
+ AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_JOB_NAME,
+ AnalyticsConstants.TCA_DMAAP_PUBLISHER_QUARTZ_TRIGGER_NAME);
+ }
+}
Code Review / dcaegen2 / analytics / tca.git / blobdiff