NIFI-1196 Providing handling of FETCH provenance events for their "unique" property, transit URI, within the framework and UI.

Reviewed by Tony Kurc (tkurc@apache.org)

nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardLineageResult.java

@@ -288,6 +288,7 @@ public class StandardLineageResult implements ComputeLineageResult {
                 }
                 break;
                 case RECEIVE:
+                case FETCH:
                 case CREATE: {
                     // for a receive event, we want to create a FlowFile Node that represents the FlowFile received
                     // and create an edge from the Receive Event to the FlowFile Node

nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java

@@ -728,6 +728,7 @@ public final class StandardProvenanceEventRecord implements ProvenanceEventRecor
                     }
                     break;
                 case RECEIVE:
+                case FETCH:
                 case SEND:
                     assertSet(transitUri, "Transit URI");
                     break;

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java

@@ -457,6 +457,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
                     bytesSent += event.getFileSize();
                     break;
                 case RECEIVE:
+                case FETCH:
                     flowFilesReceived++;
                     bytesReceived += event.getFileSize();
                     break;
@@ -616,7 +617,8 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
                     if (registeredTypes.contains(ProvenanceEventType.CREATE)
                         || registeredTypes.contains(ProvenanceEventType.FORK)
                         || registeredTypes.contains(ProvenanceEventType.JOIN)
-                        || registeredTypes.contains(ProvenanceEventType.RECEIVE)) {
+                        || registeredTypes.contains(ProvenanceEventType.RECEIVE)
+                        || registeredTypes.contains(ProvenanceEventType.FETCH)) {
                         creationEventRegistered = true;
                     }
                 }

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/webapp/js/nf/provenance/nf-provenance-table.js

@@ -1203,6 +1203,11 @@ nf.ProvenanceTable = (function () {
                 formatEventDetail('Relationship', event.relationship);
             }
 
+            // conditionally show FETCH details
+            if (event.eventType === 'FETCH') {
+                formatEventDetail('Transit Uri', event.transitUri);
+            }
+
             // conditionally show the cluster node identifier
             if (nf.Common.isDefinedAndNotNull(event.clusterNodeId)) {
                 // save the cluster node id

nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordReader.java

@@ -347,6 +347,8 @@ public class StandardRecordReader implements RecordReader {
         } else if (eventType == ProvenanceEventType.RECEIVE) {
             builder.setTransitUri(readNullableString(dis));
             builder.setSourceSystemFlowFileIdentifier(readNullableString(dis));
+        } else if (eventType == ProvenanceEventType.FETCH) {
+            builder.setTransitUri(readNullableString(dis));
         } else if (eventType == ProvenanceEventType.SEND) {
             builder.setTransitUri(readNullableString(dis));
         } else if (eventType == ProvenanceEventType.ADDINFO) {

nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/StandardRecordWriter.java

@@ -235,6 +235,8 @@ public class StandardRecordWriter implements RecordWriter {
             } else if (recordType == ProvenanceEventType.RECEIVE) {
                 writeNullableString(out, record.getTransitUri());
                 writeNullableString(out, record.getSourceSystemFlowFileIdentifier());
+            } else if (recordType == ProvenanceEventType.FETCH) {
+                writeNullableString(out, record.getTransitUri());
             } else if (recordType == ProvenanceEventType.SEND) {
                 writeNullableString(out, record.getTransitUri());
             } else if (recordType == ProvenanceEventType.ADDINFO) {