throttle job finished notifications for streaming feeds
@@ -41,6 +41,7 @@ import com.thinkbiganalytics.nifi.rest.client.LegacyNifiRestClient;
import org.apache.nifi.web.api.dto.BulletinDTO;
import org.hibernate.exception.LockAcquisitionException;
+import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
@@ -72,8 +73,21 @@ public class ProvenanceEventReceiver implements FailedStepExecutionListener {
Cache<String, String> completedJobEvents = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).build();
+ /**
+ * Temporary cache of completed events in to check against to ensure we trigger the same event twice
+ */
+ Cache<String, DateTime> lastFeedFinishedNotificationCache = CacheBuilder.newBuilder().build();
+
+
+ /**
+ * Should ops manager query NiFi for any related bulletins on failure events
+ */
@Value("${kylo.ops.mgr.query.nifi.bulletins:false}")
private boolean queryForNiFiBulletins;
+
+ @Value("${kylo.ops.mgr.stream.finished.notification.wait-time-sec:5}")
+ private Integer streamJobFinishedNotificationWaitTime = 5;
+
@Inject
private NifiEventProvider nifiEventProvider;
@Inject
@@ -264,23 +278,43 @@ public class ProvenanceEventReceiver implements FailedStepExecutionListener {
}
}
+ /**
+ * When should the system notify others the job has finished with success/failure.
+ * if Batch always notify
+ * if Stream notify every 5 sec
+ * @param event final event
+ * @return true if notify, false if not
+ */
+ private boolean isNotifyJobFinished(ProvenanceEventRecordDTO event){
+ if (event.isFinalJobEvent()) {
+ String mapKey = triggeredEventsKey(event);
+ String alreadyTriggered = completedJobEvents.getIfPresent(mapKey);
+ if (alreadyTriggered == null) {
+ return !event.isStream() ||
+ (lastFeedFinishedNotificationCache.getIfPresent(event.getFeedName()) == null ||
+ DateTime.now().minusSeconds(streamJobFinishedNotificationWaitTime).isAfter(lastFeedFinishedNotificationCache.getIfPresent(event.getFeedName())));
+ }
+ }
+ return false;
+
+}
/**
* Notify that the Job is complete either as a successful job or failed Job
+ * if its a streaming event notifications will go out every xx seconds default 5
+ * if its a batch it will always notify
*
* @param event a provenance event
*/
private void notifyJobFinished(BatchJobExecution jobExecution, ProvenanceEventRecordDTO event) {
- if (event.isFinalJobEvent()) {
+ if (isNotifyJobFinished(event)) {
+ //register the event as being triggered
String mapKey = triggeredEventsKey(event);
- String alreadyTriggered = completedJobEvents.getIfPresent(mapKey);
- if (alreadyTriggered == null) {
completedJobEvents.put(mapKey, mapKey);
- /// TRIGGER JOB COMPLETE!!!
- //TODO if Stream do we just notify every x sec?
+ lastFeedFinishedNotificationCache.put(event.getFeedName(),DateTime.now());
metadataAccess.commit(() -> {
BatchJobExecution batchJobExecution = jobExecution;
- if (batchJobExecution.isFailed()) {
+ if ((!event.isStream() && batchJobExecution.isFailed()) || (event.isStream() && event.isFailure())) {
//requery for failure events as we need to access the map of data for alert generation
batchJobExecution = batchJobExecutionProvider.findByJobExecutionId(jobExecution.getJobExecutionId());
failedJob(batchJobExecution, event);
@@ -289,7 +323,8 @@ public class ProvenanceEventReceiver implements FailedStepExecutionListener {
}
}, MetadataAccess.SERVICE);
- }
+ }else {
+ log.debug("skipping job finished notification for feed: {}, isStream:{}, isFailure:{} ",event.getFeedName(),event.isStream(), event.isFailure());
}
}
