You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2014/12/16 17:21:59 UTC
[04/27] incubator-nifi git commit: NIFI-72: Auto-generate CREATE,
CONTENT_MODIFIED, ATTRIBUTES_MODIFIED events when appropriate
NIFI-72: Auto-generate CREATE, CONTENT_MODIFIED, ATTRIBUTES_MODIFIED events when appropriate
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/6b0a5e8c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/6b0a5e8c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/6b0a5e8c
Branch: refs/heads/nifi-27
Commit: 6b0a5e8cd75a03a13ca90b08204bccc2a45bee70
Parents: 8ed8d69
Author: Mark Payne <ma...@hotmail.com>
Authored: Thu Dec 11 09:03:05 2014 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Dec 11 09:03:05 2014 -0500
----------------------------------------------------------------------
.../repository/StandardProcessSession.java | 83 +++++++-
.../repository/StandardProvenanceReporter.java | 4 +-
.../repository/TestStandardProcessSession.java | 211 +++++++++++++------
3 files changed, 232 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6b0a5e8c/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java b/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 08e6afe..4ba45aa 100644
--- a/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++ b/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -75,7 +75,6 @@ import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.ProvenanceReporter;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
import org.apache.nifi.util.NiFiProperties;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -488,6 +487,16 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
}
+ private void addEventType(final Map<String, Set<ProvenanceEventType>> map, final String id, final ProvenanceEventType eventType) {
+ Set<ProvenanceEventType> eventTypes = map.get(id);
+ if ( eventTypes == null ) {
+ eventTypes = new HashSet<>();
+ map.put(id, eventTypes);
+ }
+
+ eventTypes.add(eventType);
+ }
+
private void updateProvenanceRepo(final Checkpoint checkpoint) {
// Update Provenance Repository
final ProvenanceEventRepository provenanceRepo = context.getProvenanceRepository();
@@ -496,7 +505,8 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
// in case the Processor developer submitted the same events to the reporter. So we use a LinkedHashSet
// for this, so that we are able to ensure that the events are submitted in the proper order.
final Set<ProvenanceEventRecord> recordsToSubmit = new LinkedHashSet<>();
-
+ final Map<String, Set<ProvenanceEventType>> eventTypesPerFlowFileId = new HashMap<>();
+
final Set<ProvenanceEventRecord> processorGenerated = checkpoint.reportedEvents;
// We first want to submit FORK events because if the Processor is going to create events against
@@ -513,6 +523,13 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
if (!event.getChildUuids().isEmpty() && !isSpuriousForkEvent(event, checkpoint.removedFlowFiles) && !processorGenerated.contains(event)) {
recordsToSubmit.add(event);
+
+ for ( final String childUuid : event.getChildUuids() ) {
+ addEventType(eventTypesPerFlowFileId, childUuid, event.getEventType());
+ }
+ for ( final String parentUuid : event.getParentUuids() ) {
+ addEventType(eventTypesPerFlowFileId, parentUuid, event.getEventType());
+ }
}
}
@@ -523,6 +540,7 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
recordsToSubmit.add(event);
+ addEventType(eventTypesPerFlowFileId, event.getFlowFileUuid(), event.getEventType());
}
// Finally, add any other events that we may have generated.
@@ -533,6 +551,67 @@ public final class StandardProcessSession implements ProcessSession, ProvenanceE
}
recordsToSubmit.add(event);
+ addEventType(eventTypesPerFlowFileId, event.getFlowFileUuid(), event.getEventType());
+ }
+ }
+
+ // Check if content or attributes changed. If so, register the appropriate events.
+ for (final StandardRepositoryRecord repoRecord : checkpoint.records.values() ) {
+ final ContentClaim original = repoRecord.getOriginalClaim();
+ final ContentClaim current = repoRecord.getCurrentClaim();
+
+ boolean contentChanged = false;
+ if ( original == null && current != null ) {
+ contentChanged = true;
+ }
+ if ( original != null && current == null ) {
+ contentChanged = true;
+ }
+ if ( original != null && current != null && !original.equals(current) ) {
+ contentChanged = true;
+ }
+
+ final FlowFileRecord curFlowFile = repoRecord.getCurrent();
+ final String flowFileId = curFlowFile.getAttribute(CoreAttributes.UUID.key());
+ boolean eventAdded = false;
+
+ if (checkpoint.removedFlowFiles.contains(flowFileId)) {
+ continue;
+ }
+
+ if ( contentChanged ) {
+ recordsToSubmit.add(provenanceReporter.build(curFlowFile, ProvenanceEventType.CONTENT_MODIFIED).build());
+ addEventType(eventTypesPerFlowFileId, flowFileId, ProvenanceEventType.CONTENT_MODIFIED);
+ eventAdded = true;
+ }
+
+ if ( checkpoint.createdFlowFiles.contains(flowFileId) ) {
+ final Set<ProvenanceEventType> registeredTypes = eventTypesPerFlowFileId.get(flowFileId);
+ boolean creationEventRegistered = false;
+ if ( registeredTypes != null ) {
+ if ( registeredTypes.contains(ProvenanceEventType.CREATE) ||
+ registeredTypes.contains(ProvenanceEventType.FORK) ||
+ registeredTypes.contains(ProvenanceEventType.JOIN) ||
+ registeredTypes.contains(ProvenanceEventType.RECEIVE) ) {
+ creationEventRegistered = true;
+ }
+ }
+
+ if ( !creationEventRegistered ) {
+ recordsToSubmit.add(provenanceReporter.build(curFlowFile, ProvenanceEventType.CREATE).build());
+ eventAdded = true;
+ }
+ }
+
+ if ( !eventAdded && !repoRecord.getUpdatedAttributes().isEmpty() ) {
+ // We generate an ATTRIBUTES_MODIFIED event only if no other event has been
+ // created for the FlowFile. We do this because all events contain both the
+ // newest and the original attributes, so generating an ATTRIBUTES_MODIFIED
+ // event is redundant if another already exists.
+ if ( !eventTypesPerFlowFileId.containsKey(flowFileId) ) {
+ recordsToSubmit.add(provenanceReporter.build(curFlowFile, ProvenanceEventType.ATTRIBUTES_MODIFIED).build());
+ addEventType(eventTypesPerFlowFileId, flowFileId, ProvenanceEventType.ATTRIBUTES_MODIFIED);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6b0a5e8c/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java b/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
index e8b1e87..01fb3dc 100644
--- a/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
+++ b/nar-bundles/framework-bundle/framework/core/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
@@ -328,7 +328,7 @@ public class StandardProvenanceReporter implements ProvenanceReporter {
}
}
}
-
+
@Override
public void modifyContent(final FlowFile flowFile) {
modifyContent(flowFile, null, -1L);
@@ -421,7 +421,7 @@ public class StandardProvenanceReporter implements ProvenanceReporter {
}
}
- private ProvenanceEventBuilder build(final FlowFile flowFile, final ProvenanceEventType eventType) {
+ ProvenanceEventBuilder build(final FlowFile flowFile, final ProvenanceEventType eventType) {
final ProvenanceEventBuilder builder = repository.eventBuilder();
builder.setEventType(eventType);
builder.fromFlowFile(flowFile);
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6b0a5e8c/nar-bundles/framework-bundle/framework/core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
----------------------------------------------------------------------
diff --git a/nar-bundles/framework-bundle/framework/core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java b/nar-bundles/framework-bundle/framework/core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
index 6e0a5d7..3dbbcf3 100644
--- a/nar-bundles/framework-bundle/framework/core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
+++ b/nar-bundles/framework-bundle/framework/core/src/test/java/org/apache/nifi/controller/repository/TestStandardProcessSession.java
@@ -17,6 +17,7 @@
package org.apache.nifi.controller.repository;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;
@@ -64,7 +65,6 @@ import org.apache.nifi.provenance.MockProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
-
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -267,7 +267,7 @@ public class TestStandardProcessSession {
}
@Test
- public void testSpawnsNotEmittedIfFilesDeleted() throws IOException {
+ public void testForksNotEmittedIfFilesDeleted() throws IOException {
final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
.addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
.entryDate(System.currentTimeMillis())
@@ -283,8 +283,9 @@ public class TestStandardProcessSession {
assertEquals(0, provenanceRepo.getEvents(0L, 100000).size());
}
+
@Test
- public void testProvenanceEventsEmittedForSpawnIfNotRemoved() throws IOException {
+ public void testProvenanceEventsEmittedForForkIfNotRemoved() throws IOException {
final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
.addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
.entryDate(System.currentTimeMillis())
@@ -320,6 +321,79 @@ public class TestStandardProcessSession {
}
@Test
+ public void testUpdateAttributesThenJoin() throws IOException {
+ final FlowFileRecord flowFileRecord1 = new StandardFlowFileRecord.Builder()
+ .id(1L)
+ .addAttribute("uuid", "11111111-1111-1111-1111-111111111111")
+ .entryDate(System.currentTimeMillis())
+ .build();
+
+ final FlowFileRecord flowFileRecord2 = new StandardFlowFileRecord.Builder()
+ .id(2L)
+ .addAttribute("uuid", "22222222-2222-2222-2222-222222222222")
+ .entryDate(System.currentTimeMillis())
+ .build();
+
+ flowFileQueue.put(flowFileRecord1);
+ flowFileQueue.put(flowFileRecord2);
+
+ FlowFile ff1 = session.get();
+ FlowFile ff2 = session.get();
+
+ ff1 = session.putAttribute(ff1, "index", "1");
+ ff2 = session.putAttribute(ff2, "index", "2");
+
+ final List<FlowFile> parents = new ArrayList<>(2);
+ parents.add(ff1);
+ parents.add(ff2);
+
+ final FlowFile child = session.create(parents);
+
+ final Relationship rel = new Relationship.Builder().name("A").build();
+
+ session.transfer(ff1, rel);
+ session.transfer(ff2, rel);
+ session.transfer(child, rel);
+
+ session.commit();
+
+ final List<ProvenanceEventRecord> events = provenanceRepo.getEvents(0L, 1000);
+
+ // We should have a JOIN and 2 ATTRIBUTE_MODIFIED's
+ assertEquals(3, events.size());
+
+ int joinCount = 0;
+ int ff1UpdateCount = 0;
+ int ff2UpdateCount = 0;
+
+ for ( final ProvenanceEventRecord event : events ) {
+ switch (event.getEventType()) {
+ case JOIN:
+ assertEquals(child.getAttribute("uuid"), event.getFlowFileUuid());
+ joinCount++;
+ break;
+ case ATTRIBUTES_MODIFIED:
+ if ( event.getFlowFileUuid().equals(ff1.getAttribute("uuid")) ) {
+ ff1UpdateCount++;
+ } else if ( event.getFlowFileUuid().equals(ff2.getAttribute("uuid")) ) {
+ ff2UpdateCount++;
+ } else {
+ Assert.fail("Got ATTRIBUTE_MODIFIED for wrong FlowFile: " + event.getFlowFileUuid());
+ }
+ break;
+ default:
+ Assert.fail("Unexpected event type: " + event);
+ }
+ }
+
+ assertEquals(1, joinCount);
+ assertEquals(1, ff1UpdateCount);
+ assertEquals(1, ff2UpdateCount);
+
+ assertEquals(1, joinCount);
+ }
+
+ @Test
public void testForkOneToOneReported() throws IOException {
final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
.addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
@@ -628,34 +702,34 @@ public class TestStandardProcessSession {
@Test
public void testContentNotFoundExceptionThrownWhenUnableToReadDataOffsetTooLarge() {
final FlowFileRecord flowFileRecord = new StandardFlowFileRecord.Builder()
- .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
- .entryDate(System.currentTimeMillis())
- .contentClaim(new ContentClaim() {
- @Override
- public int compareTo(ContentClaim arg0) {
- return 0;
- }
-
- @Override
- public String getId() {
- return "0";
- }
-
- @Override
- public String getContainer() {
- return "container";
- }
-
- @Override
- public String getSection() {
- return "section";
- }
-
- @Override
- public boolean isLossTolerant() {
- return true;
- }
- }).build();
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .contentClaim(new ContentClaim() {
+ @Override
+ public int compareTo(ContentClaim arg0) {
+ return 0;
+ }
+
+ @Override
+ public String getId() {
+ return "0";
+ }
+
+ @Override
+ public String getContainer() {
+ return "container";
+ }
+
+ @Override
+ public String getSection() {
+ return "section";
+ }
+
+ @Override
+ public boolean isLossTolerant() {
+ return true;
+ }
+ }).build();
flowFileQueue.put(flowFileRecord);
FlowFile ff1 = session.get();
@@ -668,37 +742,35 @@ public class TestStandardProcessSession {
session.commit();
final FlowFileRecord flowFileRecord2 = new StandardFlowFileRecord.Builder()
- .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
- .entryDate(System.currentTimeMillis())
- .contentClaim(new ContentClaim() {
- @Override
- public int compareTo(ContentClaim arg0) {
- return 0;
- }
-
- @Override
- public String getId() {
- return "0";
- }
-
- @Override
- public String getContainer() {
- return "container";
- }
-
- @Override
- public String getSection() {
- return "section";
- }
-
- @Override
- public boolean isLossTolerant() {
- return true;
- }
- })
- .contentClaimOffset(1000L)
- .size(1L)
- .build();
+ .addAttribute("uuid", "12345678-1234-1234-1234-123456789012")
+ .entryDate(System.currentTimeMillis())
+ .contentClaim(new ContentClaim() {
+ @Override
+ public int compareTo(ContentClaim arg0) {
+ return 0;
+ }
+
+ @Override
+ public String getId() {
+ return "0";
+ }
+
+ @Override
+ public String getContainer() {
+ return "container";
+ }
+
+ @Override
+ public String getSection() {
+ return "section";
+ }
+
+ @Override
+ public boolean isLossTolerant() {
+ return true;
+ }
+ })
+ .contentClaimOffset(1000L).size(1L).build();
flowFileQueue.put(flowFileRecord2);
// attempt to read the data.
@@ -759,6 +831,21 @@ public class TestStandardProcessSession {
}
}
+
+ @Test
+ public void testCreateEmitted() throws IOException {
+ FlowFile newFlowFile = session.create();
+ session.transfer(newFlowFile, new Relationship.Builder().name("A").build());
+ session.commit();
+
+ final List<ProvenanceEventRecord> events = provenanceRepo.getEvents(0L, 10000);
+ assertFalse(events.isEmpty());
+ assertEquals(1, events.size());
+
+ final ProvenanceEventRecord event = events.get(0);
+ assertEquals(ProvenanceEventType.CREATE, event.getEventType());
+ }
+
private static class MockFlowFileRepository implements FlowFileRepository {
private final AtomicLong idGenerator = new AtomicLong(0L);