Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-9783: This closes #5855. When migrating FlowFiles from one ProcessSession to another, if any FlowFile had already been transferred, and the Relationship to which it was transferred was auto-terminated, we were updating the wrong member variable, which threw off our stats for the processor. Fixed that. 

Signed-off-by: Joe Witt <joewitt@apache.org>
This commit is contained in:
Mark Payne 2022-03-09 13:44:08 -05:00 committed by Joe Witt
parent c73573b325
commit 73356ea448
No known key found for this signature in database
GPG Key ID: 9093BF854F811A1A
2 changed files with 50 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
View File

@ -1599,11 +1599,24 @@ public class StandardProcessSession implements ProcessSession, ProvenanceEventEn
}
if (repoRecord.getTransferRelationship() != null) {
flowFilesOut--;
contentSizeOut -= flowFile.getSize();
final Relationship transferRelationship = repoRecord.getTransferRelationship();
final Collection<Connection> destinations = context.getConnections(transferRelationship);
final int numDestinations = destinations.size();
final boolean autoTerminated = numDestinations == 0 && context.getConnectable().isAutoTerminated(transferRelationship);
newOwner.flowFilesOut++;
newOwner.contentSizeOut += flowFile.getSize();
if (autoTerminated) {
removedCount--;
removedBytes -= flowFile.getSize();
newOwner.removedCount++;
newOwner.removedBytes += flowFile.getSize();
} else {
flowFilesOut--;
contentSizeOut -= flowFile.getSize();
newOwner.flowFilesOut++;
newOwner.contentSizeOut += flowFile.getSize();
}
}
final List<ProvenanceEventRecord> events = generatedProvenanceEvents.remove(flowFile);

33
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
View File

@ -2452,6 +2452,39 @@ public class StandardProcessSessionIT {
session.commit();
}
@Test
public void testMigrateAfterTransferToAutoTerminatedRelationship() {
final long start = System.currentTimeMillis();
FlowFile flowFile = session.create();
flowFile = session.write(flowFile, out -> out.write("Hello".getBytes(StandardCharsets.UTF_8)));
final StandardProcessSession newSession = new StandardProcessSession(context, () -> false);
when(connectable.getConnections(any(Relationship.class))).thenReturn(Collections.emptySet());
when(connectable.isAutoTerminated(any(Relationship.class))).thenReturn(true);
session.transfer(flowFile, new Relationship.Builder().name("success").build());
session.migrate(newSession, Collections.singleton(flowFile));
session.commit();
RepositoryStatusReport report = flowFileEventRepository.reportTransferEvents(start - 1);
FlowFileEvent event = report.getReportEntries().values().iterator().next();
assertEquals(0, event.getFlowFilesRemoved());
assertEquals(0, event.getContentSizeRemoved());
assertEquals(0, event.getFlowFilesOut());
assertEquals(0, event.getContentSizeOut());
newSession.commit();
report = flowFileEventRepository.reportTransferEvents(start - 1);
event = report.getReportEntries().values().iterator().next();
assertEquals(1, event.getFlowFilesRemoved());
assertEquals(5, event.getContentSizeRemoved());
assertEquals(0, event.getFlowFilesOut());
assertEquals(0, event.getContentSizeOut());
}
@Test
public void testNewFlowFileModifiedMultipleTimesHasTransientClaimsOnCommit() {
FlowFile flowFile = session.create();