This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 b0168a6eeda NIFI-15758 Added Fragment Attributes to FlowFile format in
UnpackContent (#11526)
b0168a6eeda is described below
commit b0168a6eedac3cdf7310a87d1aac4e49635a0e82
Author: David Handermann <[email protected]>
AuthorDate: Tue Aug 11 07:19:15 2026 -0500
NIFI-15758 Added Fragment Attributes to FlowFile format in UnpackContent
(#11526)
---
.../nifi/processors/standard/UnpackContent.java | 116 ++++++++++++---------
.../processors/standard/TestUnpackContent.java | 66 ++++++++++++
2 files changed, 133 insertions(+), 49 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
index 3100e6982fb..63bc8273d83 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java
@@ -311,32 +311,13 @@ public class UnpackContent extends AbstractProcessor {
}
}
- // set the Unpacker to use for this FlowFile. FlowFileUnpackager
objects maintain state and are not reusable.
- final Unpacker unpacker;
- final boolean addFragmentAttrs = switch (packagingFormat) {
- case TAR_FORMAT -> {
- unpacker = tarUnpacker;
- yield true;
- }
- case ZIP_FORMAT -> {
- unpacker = zipUnpacker;
- yield true;
- }
- case FLOWFILE_STREAM_FORMAT_V2 -> {
- unpacker = new FlowFileStreamUnpacker(new
FlowFileUnpackagerV2());
- yield false;
- }
- case FLOWFILE_STREAM_FORMAT_V3 -> {
- unpacker = new FlowFileStreamUnpacker(new
FlowFileUnpackagerV3());
- yield false;
- }
- case FLOWFILE_TAR_FORMAT -> {
- unpacker = new FlowFileStreamUnpacker(new
FlowFileUnpackagerV1());
- yield false;
- }
- default ->
- // The format of the unpacker should be known before
initialization
- throw new ProcessException(packagingFormat + " is not a valid
packaging format");
+ final Unpacker unpacker = switch (packagingFormat) {
+ case TAR_FORMAT -> tarUnpacker;
+ case ZIP_FORMAT -> zipUnpacker;
+ case FLOWFILE_STREAM_FORMAT_V2 -> new FlowFileStreamUnpacker(new
FlowFileUnpackagerV2());
+ case FLOWFILE_STREAM_FORMAT_V3 -> new FlowFileStreamUnpacker(new
FlowFileUnpackagerV3());
+ case FLOWFILE_TAR_FORMAT -> new FlowFileStreamUnpacker(new
FlowFileUnpackagerV1());
+ default -> throw new ProcessException("Format [%s] not
supported".formatted(packagingFormat));
};
final List<FlowFile> unpacked = new ArrayList<>();
@@ -348,9 +329,17 @@ public class UnpackContent extends AbstractProcessor {
return;
}
- if (addFragmentAttrs) {
- finishFragmentAttributes(session, flowFile, unpacked);
+ // Determine whether Fragment Identifiers are needed based on
Unpacker and existing attribute status after unpacking
+ final boolean fragmentIdentifiersRequired;
+ if (unpacker instanceof final FlowFileStreamUnpacker
streamUnpacker) {
+ fragmentIdentifiersRequired =
!streamUnpacker.isFragmentIdentifierRestored();
+ } else {
+ fragmentIdentifiersRequired = true;
}
+ if (fragmentIdentifiersRequired) {
+ putFragmentAttributes(session, flowFile, unpacked);
+ }
+
session.transfer(unpacked, REL_SUCCESS);
final String fragmentId = !unpacked.isEmpty() ?
unpacked.getFirst().getAttribute(FRAGMENT_ID) : null;
flowFile = FragmentAttributes.copyAttributesToOriginal(session,
flowFile, fragmentId, unpacked.size());
@@ -648,10 +637,16 @@ public class UnpackContent extends AbstractProcessor {
private final FlowFileUnpackager unpackager;
+ private boolean fragmentIdentifierRestored;
+
public FlowFileStreamUnpacker(final FlowFileUnpackager unpackager) {
this.unpackager = unpackager;
}
+ private boolean isFragmentIdentifierRestored() {
+ return fragmentIdentifierRestored;
+ }
+
@Override
public void unpack(final ProcessSession session, final FlowFile
source, final List<FlowFile> unpacked) {
session.read(source, inputStream -> {
@@ -677,6 +672,13 @@ public class UnpackContent extends AbstractProcessor {
// and later unpack it -- in this case, we have
two FlowFiles with the same UUID.
attributes.remove(CoreAttributes.UUID.key());
+ // Track whether the packaged FlowFile itself
carried a fragment identifier. This is evaluated
+ // against the restored attributes rather than the
unpacked FlowFile, because a child created
+ // from the source inherits the source's
attributes and could otherwise report a false positive.
+ if (attributes.containsKey(FRAGMENT_ID)) {
+ fragmentIdentifierRestored = true;
+ }
+
if
(!attributes.containsKey(CoreAttributes.MIME_TYPE.key())) {
attributes.put(CoreAttributes.MIME_TYPE.key(),
OCTET_STREAM);
}
@@ -691,33 +693,49 @@ public class UnpackContent extends AbstractProcessor {
}
}
- private void finishFragmentAttributes(final ProcessSession session, final
FlowFile source, final List<FlowFile> unpacked) {
- // first pass verifies all FlowFiles have the FRAGMENT_INDEX attribute
and gets the total number of fragments
- int fragmentCount = 0;
- for (FlowFile ff : unpacked) {
- String fragmentIndex = ff.getAttribute(FRAGMENT_INDEX);
- if (fragmentIndex != null) {
- fragmentCount++;
- } else {
- return;
+ private void putFragmentAttributes(
+ final ProcessSession session,
+ final FlowFile source,
+ final List<FlowFile> unpacked
+ ) {
+ final String fragmentId = UUID.randomUUID().toString();
+ final String segmentOriginalFilename =
getSegmentOriginalFilename(source);
+ final String fragmentCount = String.valueOf(unpacked.size());
+
+ final List<FlowFile> updated = new ArrayList<>(unpacked.size());
+ int fragmentIndex = 0;
+ for (final FlowFile unpackedFlowFile : unpacked) {
+ fragmentIndex++;
+ final Map<String, String> fragmentAttributes = new HashMap<>();
+
+ final String unpackedFragmentId =
unpackedFlowFile.getAttribute(FRAGMENT_ID);
+ if (unpackedFragmentId == null) {
+ // Set Fragment Identifier and Index when absent for FlowFile
Formats
+ fragmentAttributes.put(FRAGMENT_ID, fragmentId);
+ fragmentAttributes.put(FRAGMENT_INDEX,
String.valueOf(fragmentIndex));
}
- }
- String originalFilename =
source.getAttribute(CoreAttributes.FILENAME.key());
- if (originalFilename.endsWith(".tar") ||
originalFilename.endsWith(".zip") || originalFilename.endsWith(".pkg")) {
- originalFilename = originalFilename.substring(0,
originalFilename.length() - 4);
+ fragmentAttributes.put(FRAGMENT_COUNT, fragmentCount);
+ fragmentAttributes.put(SEGMENT_ORIGINAL_FILENAME,
segmentOriginalFilename);
+
+ updated.add(session.putAllAttributes(unpackedFlowFile,
fragmentAttributes));
}
- // second pass adds fragment attributes
- List<FlowFile> newList = new ArrayList<>(unpacked);
unpacked.clear();
- for (FlowFile ff : newList) {
- FlowFile newFF = session.putAllAttributes(ff, Map.of(
- FRAGMENT_COUNT, String.valueOf(fragmentCount),
- SEGMENT_ORIGINAL_FILENAME, originalFilename
- ));
- unpacked.add(newFF);
+ unpacked.addAll(updated);
+ }
+
+ private String getSegmentOriginalFilename(final FlowFile source) {
+ final String filename =
source.getAttribute(CoreAttributes.FILENAME.key());
+
+ final String originalFilename;
+ if (filename.endsWith(".tar") || filename.endsWith(".zip") ||
filename.endsWith(".pkg")) {
+ originalFilename = filename.substring(0, filename.length() - 4);
+ } else {
+ originalFilename = filename;
}
+
+ return originalFilename;
}
protected enum PackageFormat implements DescribedValue {
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
index 630f19d157c..80727a826b9 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestUnpackContent.java
@@ -45,12 +45,17 @@ import static
org.apache.nifi.processors.standard.SplitContent.FRAGMENT_COUNT;
import static org.apache.nifi.processors.standard.SplitContent.FRAGMENT_ID;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestUnpackContent {
private static final String FIRST_FRAGMENT_INDEX = "1";
+ private static final String EXISTING_FRAGMENT_ID = "existing-fragment-id";
+
+ private static final String EXISTING_SEGMENT_FILENAME = "original-archive";
+
private static final Path dataPath =
Paths.get("src/test/resources/TestUnpackContent");
private static final DateTimeFormatter TIMESTAMP_FORMATTER =
DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ");
@@ -451,6 +456,67 @@ public class TestUnpackContent {
}
}
+ @Test
+ public void testFlowFileStreamAssignsFragmentAttributesWhenAbsent() throws
IOException {
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.FLOWFILE_STREAM_FORMAT_V3);
+ runner.enqueue(dataPath.resolve("data.flowfilev3"));
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 2);
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+
+ final List<MockFlowFile> unpacked =
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS);
+ final String fragmentId =
unpacked.getFirst().getAttribute(UnpackContent.FRAGMENT_ID);
+ assertNotNull(fragmentId);
+ for (final MockFlowFile flowFile : unpacked) {
+ flowFile.assertAttributeEquals(UnpackContent.FRAGMENT_ID,
fragmentId);
+ flowFile.assertAttributeEquals(UnpackContent.FRAGMENT_COUNT, "2");
+
flowFile.assertAttributeEquals(UnpackContent.SEGMENT_ORIGINAL_FILENAME,
"data.flowfilev3");
+ }
+
+ unpacked.get(0).assertAttributeEquals(UnpackContent.FRAGMENT_INDEX,
FIRST_FRAGMENT_INDEX);
+ unpacked.get(1).assertAttributeEquals(UnpackContent.FRAGMENT_INDEX,
"2");
+ }
+
+ @Test
+ public void testFlowFileStreamPreservesExistingFragmentAttributes() {
+ final TestRunner mergeRunner = TestRunners.newTestRunner(new
MergeContent());
+ mergeRunner.setProperty(MergeContent.MERGE_FORMAT,
MergeContent.MergeFormat.FLOWFILE_STREAM_V3);
+ mergeRunner.setProperty(MergeContent.MERGE_STRATEGY,
MergeContent.MergeStrategy.BIN_PACK);
+ mergeRunner.setProperty(MergeContent.MIN_ENTRIES, "2");
+ mergeRunner.setProperty(MergeContent.MAX_ENTRIES, "2");
+
+ final Map<String, String> attributes = new HashMap<>();
+ attributes.put(UnpackContent.FRAGMENT_ID, EXISTING_FRAGMENT_ID);
+ attributes.put(UnpackContent.FRAGMENT_COUNT, "2");
+ attributes.put(UnpackContent.SEGMENT_ORIGINAL_FILENAME,
EXISTING_SEGMENT_FILENAME);
+ attributes.put(UnpackContent.FRAGMENT_INDEX, FIRST_FRAGMENT_INDEX);
+ mergeRunner.enqueue("Hello ".getBytes(StandardCharsets.UTF_8),
attributes);
+ attributes.put(UnpackContent.FRAGMENT_INDEX, "2");
+ mergeRunner.enqueue("World".getBytes(StandardCharsets.UTF_8),
attributes);
+ mergeRunner.run();
+
+ mergeRunner.assertTransferCount(MergeContent.REL_MERGED, 1);
+ final MockFlowFile packaged =
mergeRunner.getFlowFilesForRelationship(MergeContent.REL_MERGED).getFirst();
+
+ runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.FLOWFILE_STREAM_FORMAT_V3);
+ runner.enqueue(packaged);
+ runner.run();
+
+ runner.assertTransferCount(UnpackContent.REL_SUCCESS, 2);
+ runner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
+
+ final List<MockFlowFile> unpacked =
runner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS);
+ for (final MockFlowFile flowFile : unpacked) {
+ flowFile.assertAttributeEquals(UnpackContent.FRAGMENT_ID,
EXISTING_FRAGMENT_ID);
+ flowFile.assertAttributeEquals(UnpackContent.FRAGMENT_COUNT, "2");
+
flowFile.assertAttributeEquals(UnpackContent.SEGMENT_ORIGINAL_FILENAME,
EXISTING_SEGMENT_FILENAME);
+ }
+
+ unpacked.get(0).assertAttributeEquals(UnpackContent.FRAGMENT_INDEX,
FIRST_FRAGMENT_INDEX);
+ unpacked.get(1).assertAttributeEquals(UnpackContent.FRAGMENT_INDEX,
"2");
+ }
+
@Test
public void testTarThenMerge() throws IOException {
runner.setProperty(UnpackContent.PACKAGING_FORMAT,
UnpackContent.PackageFormat.TAR_FORMAT);
