/* * ===============================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.TxRunnable; import co.cask.cdap.api.common.Bytes; import co.cask.cdap.api.data.DatasetContext; import co.cask.cdap.api.dataset.lib.CloseableIterator; import co.cask.cdap.api.dataset.lib.KeyValue; import co.cask.cdap.api.dataset.lib.ObjectMappedTable; import co.cask.cdap.api.metrics.Metrics; import co.cask.cdap.api.worker.WorkerContext; import com.google.common.base.Joiner; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import org.apache.tephra.TransactionFailureException; import org.openecomp.dcae.apod.analytics.cdap.common.CDAPMetricsConstants; import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertEntity; import org.openecomp.dcae.apod.analytics.cdap.common.persistance.tca.TCAVESAlertsPersister; import org.openecomp.dcae.apod.analytics.cdap.tca.utils.CDAPTCAUtils; 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.DMaaPMRPublisherResponse; import org.openecomp.dcae.apod.analytics.dmaap.service.publisher.DMaaPMRPublisher; 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.util.Date; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.CDAP_ALERTS_TABLE_VARIABLE_NAME; import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_METRICS_VARIABLE_NAME; import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.DMAAP_PUBLISHER_VARIABLE_NAME; import static org.openecomp.dcae.apod.analytics.common.AnalyticsConstants.WORKER_CONTEXT_VARIABLE_NAME; /** * Quartz Job that will monitor any new alert messages in given TCA Alerts table and if any found publish them to * DMaaP MR topic *
* @author Rajiv Singla . Creation Date: 11/17/2016.
*/
@DisallowConcurrentExecution
@PersistJobDataAfterExecution
@SuppressFBWarnings("SIC_INNER_SHOULD_BE_STATIC_ANON")
public class TCADMaaPMRPublisherJob implements Job {
private static final Logger LOG = LoggerFactory.getLogger(TCADMaaPMRPublisherJob.class);
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
LOG.debug("Starting DMaaP MR Topic Publisher 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 cdapAlertsTableName = jobDataMap.getString(CDAP_ALERTS_TABLE_VARIABLE_NAME);
final WorkerContext workerContext = (WorkerContext) jobDataMap.get(WORKER_CONTEXT_VARIABLE_NAME);
final DMaaPMRPublisher publisher = (DMaaPMRPublisher) jobDataMap.get(DMAAP_PUBLISHER_VARIABLE_NAME);
final Metrics metrics = (Metrics) jobDataMap.get(DMAAP_METRICS_VARIABLE_NAME);
LOG.debug("Start looking for new message in Alerts Table: {}", cdapAlertsTableName);
// Get new alerts from alerts table
final Map