This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 2847bb87e42 NIFI-15916 Log dropped FlowFiles when content is missing
(#11221)
2847bb87e42 is described below
commit 2847bb87e423bbb03f28309c56f36e641ee56dae
Author: Alaksiej Ščarbaty <[email protected]>
AuthorDate: Tue May 12 15:24:27 2026 +0200
NIFI-15916 Log dropped FlowFiles when content is missing (#11221)
In handleContentNotFound both branches threw MissingFlowFileException
with the same generic "Unable to find content for FlowFile" message,
giving operators no FlowFile identity and no way to distinguish an
unrecoverable drop from a recoverable rollback.
Registered-claim branch (markForAbort + drop): enrich the exception
message with the FlowFile identity and emit a WARN log so the drop is
recorded even if the caller swallows the exception.
Transient-claim branch (rollback returns the FlowFile to its queue):
enrich the message only - name the in-flight content and signal the
session is rolling back so callers can tell this is recoverable.
---
.../nifi/controller/repository/StandardProcessSession.java | 5 +++--
.../nifi/controller/repository/StandardProcessSessionIT.java | 10 ++++++++--
2 files changed, 11 insertions(+), 4 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 4ae94c47ca6..a71d2d6e104 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -3751,13 +3751,14 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
if (missingClaim == registeredClaim) {
suspectRecord.markForAbort();
+ LOG.warn("Unable to find content for {}; dropping FlowFile",
suspectRecord.getCurrent(), nfe);
rollback();
- throw new MissingFlowFileException("Unable to find content for
FlowFile", nfe);
+ throw new MissingFlowFileException("Unable to find content for " +
suspectRecord.getCurrent() + "; dropping FlowFile", nfe);
}
if (missingClaim == transientClaim) {
rollback();
- throw new MissingFlowFileException("Unable to find content for
FlowFile", nfe);
+ throw new MissingFlowFileException("Unable to find in-flight
content for " + suspectRecord.getCurrent() + "; rolling back", nfe);
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
index 314e3295ec5..db119a1db7c 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionIT.java
@@ -1601,7 +1601,10 @@ public class StandardProcessSessionIT {
// attempt to read the data.
final FlowFile ff1 = session.get();
- assertThrows(MissingFlowFileException.class, () -> session.read(ff1,
InputStream::read));
+ final MissingFlowFileException ex =
assertThrows(MissingFlowFileException.class,
+ () -> session.read(ff1, InputStream::read));
+
assertTrue(ex.getMessage().contains("12345678-1234-1234-1234-123456789012"));
+ assertTrue(ex.getMessage().contains("rolling back"));
}
@Test
@@ -1775,7 +1778,10 @@ public class StandardProcessSessionIT {
session.get();
final FlowFile ff2 = session.get();
- assertThrows(MissingFlowFileException.class, () -> session.read(ff2,
InputStream::read));
+ final MissingFlowFileException ex =
assertThrows(MissingFlowFileException.class,
+ () -> session.read(ff2, InputStream::read));
+
assertTrue(ex.getMessage().contains("12345678-1234-1234-1234-123456789012"));
+ assertTrue(ex.getMessage().contains("rolling back"));
}
@Test
