Streaming feeds always Initial/stopped
@@ -39,6 +39,7 @@ import com.thinkbiganalytics.metadata.api.jobrepo.nifi.NifiFeedStatisticsProvide
import com.thinkbiganalytics.metadata.api.jobrepo.nifi.NifiFeedStats;
import com.thinkbiganalytics.metadata.jpa.jobrepo.nifi.JpaNifiFeedProcessorStats;
import com.thinkbiganalytics.metadata.jpa.jobrepo.nifi.JpaNifiFeedStats;
+import com.thinkbiganalytics.nifi.provenance.model.ProvenanceEventRecordDTO;
import com.thinkbiganalytics.nifi.provenance.model.ProvenanceEventRecordDTOHolder;
import com.thinkbiganalytics.nifi.provenance.model.stats.AggregatedFeedProcessorStatistics;
import com.thinkbiganalytics.nifi.provenance.model.stats.AggregatedFeedProcessorStatisticsHolder;
@@ -118,6 +119,9 @@ public class NifiStatsJmsReceiver implements ClusterServiceMessageReceiver {
@Inject
private ClusterService clusterService;
+ @Inject
+ private ProvenanceEventReceiver provenanceEventReceiver;
+
@Value("${kylo.ops.mgr.stats.nifi.bulletins.mem.size:30}")
private Integer errorsToStorePerFeed = 30;
@@ -271,6 +275,7 @@ public class NifiStatsJmsReceiver implements ClusterServiceMessageReceiver {
//offload the query to nifi and merge back in
failedStatsWithFlowFiles.add((JpaNifiFeedProcessorStats) savedStats);
}
+ ensureStreamingJobExecutionRecord(stat);
}
if (stats instanceof AggregatedFeedProcessorStatisticsHolderV2) {
saveFeedStats((AggregatedFeedProcessorStatisticsHolderV2) stats, summaryStats);
@@ -496,6 +501,36 @@ public class NifiStatsJmsReceiver implements ClusterServiceMessageReceiver {
}
+ private void ensureStreamingJobExecutionRecord(NifiFeedProcessorStats stats) {
+ if(stats.getJobsStarted() >0 || stats.getJobsFinished() >0) {
+ OpsManagerFeed feed = provenanceEventFeedUtil.getFeed(stats.getFeedName());
+ if(feed.isStream()) {
+ ProvenanceEventRecordDTO event = new ProvenanceEventRecordDTO();
+ event.setEventId(stats.getMaxEventId());
+ event.setEventTime(stats.getMinEventTime().getMillis());
+ event.setEventDuration(stats.getDuration());
+ event.setFlowFileUuid(stats.getLatestFlowFileId());
+ event.setJobFlowFileId(stats.getLatestFlowFileId());
+ event.setComponentId(stats.getProcessorId());
+ event.setComponentName(stats.getProcessorName());
+ event.setIsFailure(stats.getFailedCount() > 0L);
+ event.setStream(feed.isStream());
+ event.setIsStartOfJob(stats.getJobsStarted() > 0L);
+ event.setIsFinalJobEvent(stats.getJobsFinished() > 0L);
+ event.setFeedProcessGroupId(stats.getFeedProcessGroupId());
+ event.setFeedName(stats.getFeedName());
+ ProvenanceEventRecordDTOHolder holder = new ProvenanceEventRecordDTOHolder();
+ List<ProvenanceEventRecordDTO> events = new ArrayList<>();
+ events.add(event);
+ holder.setEvents(events);
+ log.info("Ensuring Streaming Feed Event: {} has a respective JobExecution record ",event);
+ provenanceEventReceiver.receiveEvents(holder);
+ }
+ }
+
+
+ }
+
private NifiFeedProcessorStats toSummaryStats(GroupedStats groupedStats) {
NifiFeedProcessorStats nifiFeedProcessorStats = new JpaNifiFeedProcessorStats();
