You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2024/08/16 20:58:10 UTC
(nifi) branch main updated: NIFI-12709 Added New FlowFlow Attributes for Zip and Tar in UnpackContent
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory 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 9d51c675cb NIFI-12709 Added New FlowFlow Attributes for Zip and Tar in UnpackContent
9d51c675cb is described below
commit 9d51c675cbfd20e6ecbf2660afd2250ed8752f69
Author: dan-s1 <ds...@gmail.com>
AuthorDate: Fri Jul 26 17:10:22 2024 +0000
NIFI-12709 Added New FlowFlow Attributes for Zip and Tar in UnpackContent
This closes #9122
Signed-off-by: David Handermann <ex...@apache.org>
---
.../nifi/processors/standard/UnpackContent.java | 138 ++++++++++++++++-----
.../processors/standard/TestUnpackContent.java | 38 ++++--
2 files changed, 137 insertions(+), 39 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 492c488114..b831751b6e 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
@@ -67,9 +67,12 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.nio.file.Path;
+import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -99,12 +102,17 @@ import java.util.regex.Pattern;
@WritesAttribute(attribute = "fragment.count", description = "The number of unpacked FlowFiles generated from the parent FlowFile"),
@WritesAttribute(attribute = "segment.original.filename ", description = "The filename of the parent FlowFile. Extensions of .tar, .zip or .pkg are removed because "
+ "the MergeContent processor automatically adds those extensions if it is used to rebuild the original FlowFile"),
- @WritesAttribute(attribute = "file.lastModifiedTime", description = "The date and time that the unpacked file was last modified (tar only)."),
- @WritesAttribute(attribute = "file.creationTime", description = "The date and time that the file was created. This attribute holds always the same value as file.lastModifiedTime (tar only)."),
- @WritesAttribute(attribute = "file.owner", description = "The owner of the unpacked file (tar only)"),
- @WritesAttribute(attribute = "file.group", description = "The group owner of the unpacked file (tar only)"),
- @WritesAttribute(attribute = "file.permissions", description = "The read/write/execute permissions of the unpacked file (tar only)"),
- @WritesAttribute(attribute = "file.encryptionMethod", description = "The encryption method for entries in Zip archives")})
+ @WritesAttribute(attribute = UnpackContent.FILE_LAST_MODIFIED_TIME_ATTRIBUTE, description = "The date and time that the unpacked file was last modified (tar and zip only)."),
+ @WritesAttribute(attribute = UnpackContent.FILE_CREATION_TIME_ATTRIBUTE, description = "The date and time that the file was created. For encrypted zip files this attribute" +
+ " always holds the same value as " + UnpackContent.FILE_LAST_MODIFIED_TIME_ATTRIBUTE + ". For tar and unencrypted zip files if available it will be returned otherwise" +
+ " this will be the same value as" + UnpackContent.FILE_LAST_MODIFIED_TIME_ATTRIBUTE + "."),
+ @WritesAttribute(attribute = UnpackContent.FILE_LAST_METADATA_CHANGE_ATTRIBUTE, description = "The date and time the file's metadata changed (tar only)."),
+ @WritesAttribute(attribute = UnpackContent.FILE_LAST_ACCESS_TIME_ATTRIBUTE, description = "The date and time the file was last accessed (tar and unencrypted zip files only)"),
+ @WritesAttribute(attribute = UnpackContent.FILE_OWNER_ATTRIBUTE, description = "The owner of the unpacked file (tar only)"),
+ @WritesAttribute(attribute = UnpackContent.FILE_GROUP_ATTRIBUTE, description = "The group owner of the unpacked file (tar only)"),
+ @WritesAttribute(attribute = UnpackContent.FILE_SIZE_ATTRIBUTE, description = "The uncompressed size of the unpacked file (tar and zip only)"),
+ @WritesAttribute(attribute = UnpackContent.FILE_PERMISSIONS_ATTRIBUTE, description = "The read/write/execute permissions of the unpacked file (tar and unencrypted zip files only)"),
+ @WritesAttribute(attribute = UnpackContent.FILE_ENCRYPTION_METHOD_ATTRIBUTE, description = "The encryption method for entries in Zip archives")})
@SeeAlso(MergeContent.class)
@UseCase(
description = "Unpack Zip containing filenames with special characters, created on Windows with filename charset 'Cp437' or 'IBM437'.",
@@ -131,8 +139,11 @@ public class UnpackContent extends AbstractProcessor {
public static final String FILE_LAST_MODIFIED_TIME_ATTRIBUTE = "file.lastModifiedTime";
public static final String FILE_CREATION_TIME_ATTRIBUTE = "file.creationTime";
+ public static final String FILE_LAST_METADATA_CHANGE_ATTRIBUTE = "file.lastMetadataChange";
+ public static final String FILE_LAST_ACCESS_TIME_ATTRIBUTE = "file.lastAccessTime";
public static final String FILE_OWNER_ATTRIBUTE = "file.owner";
public static final String FILE_GROUP_ATTRIBUTE = "file.group";
+ public static final String FILE_SIZE_ATTRIBUTE = "file.size";
public static final String FILE_PERMISSIONS_ATTRIBUTE = "file.permissions";
public static final String FILE_ENCRYPTION_METHOD_ATTRIBUTE = "file.encryptionMethod";
@@ -373,6 +384,7 @@ public class UnpackContent extends AbstractProcessor {
@Override
public void unpack(final ProcessSession session, final FlowFile source, final List<FlowFile> unpacked) {
final String fragmentId = UUID.randomUUID().toString();
+ final Map<String, String> attributes = new HashMap<>();
session.read(source, inputStream -> {
int fragmentCount = 0;
try (final TarArchiveInputStream tarIn = new TarArchiveInputStream(new BufferedInputStream(inputStream))) {
@@ -387,26 +399,41 @@ public class UnpackContent extends AbstractProcessor {
FlowFile unpackedFile = session.create(source);
try {
- final String timeAsString = DATE_TIME_FORMATTER.format(tarEntry.getModTime().toInstant());
+ attributes.put(CoreAttributes.FILENAME.key(), file.getName());
+ attributes.put(CoreAttributes.PATH.key(), filePathString);
+ attributes.put(CoreAttributes.MIME_TYPE.key(), OCTET_STREAM);
+ attributes.put(FILE_PERMISSIONS_ATTRIBUTE, FileInfo.permissionToString(tarEntry.getMode()));
+ attributes.put(FILE_OWNER_ATTRIBUTE, String.valueOf(tarEntry.getUserName()));
+ attributes.put(FILE_GROUP_ATTRIBUTE, String.valueOf(tarEntry.getGroupName()));
+ attributes.put(FILE_SIZE_ATTRIBUTE, String.valueOf(tarEntry.getRealSize()));
+ String lastModified = DATE_TIME_FORMATTER.format(tarEntry.getModTime().toInstant());
+ attributes.put(FILE_LAST_MODIFIED_TIME_ATTRIBUTE, lastModified);
+
+ if (tarEntry.getCreationTime() != null) {
+ final String creationTime = DATE_TIME_FORMATTER.format(tarEntry.getCreationTime().toInstant());
+ attributes.put(FILE_CREATION_TIME_ATTRIBUTE, creationTime);
+ } else {
+ attributes.put(FILE_CREATION_TIME_ATTRIBUTE, lastModified);
+ }
- unpackedFile = session.putAllAttributes(unpackedFile, Map.of(
- CoreAttributes.FILENAME.key(), file.getName(),
- CoreAttributes.PATH.key(), filePathString,
- CoreAttributes.MIME_TYPE.key(), OCTET_STREAM,
+ if (tarEntry.getStatusChangeTime() != null) {
+ final String metadataChangeTime = DATE_TIME_FORMATTER.format(tarEntry.getStatusChangeTime().toInstant());
+ attributes.put(FILE_LAST_METADATA_CHANGE_ATTRIBUTE, metadataChangeTime);
+ }
- FILE_PERMISSIONS_ATTRIBUTE, FileInfo.permissionToString(tarEntry.getMode()),
- FILE_OWNER_ATTRIBUTE, String.valueOf(tarEntry.getUserName()),
- FILE_GROUP_ATTRIBUTE, String.valueOf(tarEntry.getGroupName()),
+ if (tarEntry.getLastAccessTime() != null) {
+ final String lastAccesTime = DATE_TIME_FORMATTER.format(tarEntry.getLastAccessTime().toInstant());
+ attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, lastAccesTime);
+ }
- FILE_LAST_MODIFIED_TIME_ATTRIBUTE, timeAsString,
- FILE_CREATION_TIME_ATTRIBUTE, timeAsString,
+ attributes.put(FRAGMENT_ID, fragmentId);
+ attributes.put(FRAGMENT_INDEX, String.valueOf(++fragmentCount));
- FRAGMENT_ID, fragmentId,
- FRAGMENT_INDEX, String.valueOf(++fragmentCount)
- ));
+ unpackedFile = session.putAllAttributes(unpackedFile, attributes);
final long fileSize = tarEntry.getSize();
unpackedFile = session.write(unpackedFile, outputStream -> StreamUtils.copy(tarIn, outputStream, fileSize));
+ attributes.clear();
} finally {
unpacked.add(unpackedFile);
}
@@ -470,28 +497,59 @@ public class UnpackContent extends AbstractProcessor {
return !directory && (fileFilter == null || fileFilter.matcher(fileName).find());
}
- protected void processEntry(final InputStream zipInputStream, final boolean directory, final String zipEntryName, final EncryptionMethod encryptionMethod) {
+ protected void processEntry(final InputStream zipInputStream, boolean directory, String zipEntryName, Map<String, String> attributes) {
if (isFileEntryMatched(directory, zipEntryName)) {
final File file = new File(zipEntryName);
final String parentDirectory = (file.getParent() == null) ? PATH_SEPARATOR : file.getParent();
FlowFile unpackedFile = session.create(sourceFlowFile);
try {
- unpackedFile = session.putAllAttributes(unpackedFile, Map.of(
- CoreAttributes.FILENAME.key(), file.getName(),
- CoreAttributes.PATH.key(), parentDirectory,
- CoreAttributes.MIME_TYPE.key(), OCTET_STREAM,
- FILE_ENCRYPTION_METHOD_ATTRIBUTE, encryptionMethod.toString(),
-
- FRAGMENT_ID, fragmentId,
- FRAGMENT_INDEX, String.valueOf(++fragmentIndex)
- ));
+ attributes.put(CoreAttributes.FILENAME.key(), file.getName());
+ attributes.put(CoreAttributes.PATH.key(), parentDirectory);
+ attributes.put(CoreAttributes.MIME_TYPE.key(), OCTET_STREAM);
+ attributes.put(FRAGMENT_ID, fragmentId);
+ attributes.put(FRAGMENT_INDEX, String.valueOf(++fragmentIndex));
+ unpackedFile = session.putAllAttributes(unpackedFile, attributes);
unpackedFile = session.write(unpackedFile, outputStream -> StreamUtils.copy(zipInputStream, outputStream));
} finally {
unpacked.add(unpackedFile);
}
}
}
+
+ protected void addFileSizeAttribute(long fileSize, Map<String, String> attributes) {
+ attributes.put(FILE_SIZE_ATTRIBUTE, String.valueOf(fileSize));
+ }
+
+ protected void addEncryptionMethodAttribute(EncryptionMethod encryptionMethod, Map<String, String> attributes) {
+ attributes.put(FILE_ENCRYPTION_METHOD_ATTRIBUTE, encryptionMethod.toString());
+ }
+
+ protected void addFilePermissionsAttribute(int mode, Map<String, String> attributes) {
+ if (mode > -1) {
+ attributes.put(FILE_PERMISSIONS_ATTRIBUTE, FileInfo.permissionToString(mode));
+ }
+ }
+
+ protected void addZipEntryTimeAttributes(Instant lastModified, Instant creation, Instant lastAccess, Map<String, String> attributes) {
+ String lastModifiedDate = null;
+ if (lastModified != null) {
+ lastModifiedDate = DATE_TIME_FORMATTER.format(lastModified);
+ attributes.put(FILE_LAST_MODIFIED_TIME_ATTRIBUTE, lastModifiedDate);
+ }
+
+ if (creation != null) {
+ final String creationTime = DATE_TIME_FORMATTER.format(creation);
+ attributes.put(FILE_CREATION_TIME_ATTRIBUTE, creationTime);
+ } else if (lastModifiedDate != null) {
+ attributes.put(FILE_CREATION_TIME_ATTRIBUTE, lastModifiedDate);
+ }
+
+ if (lastAccess != null) {
+ final String lastAccessDate = DATE_TIME_FORMATTER.format(lastAccess);
+ attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, lastAccessDate);
+ }
+ }
}
private static class CompressedZipInputStreamCallback extends ZipInputStreamCallback {
@@ -518,8 +576,19 @@ public class UnpackContent extends AbstractProcessor {
try (final ZipArchiveInputStream zipInputStream = new ZipArchiveInputStream(new BufferedInputStream(inputStream),
filenameEncoding.toString(), true, allowStoredEntriesWithDataDescriptor)) {
ZipArchiveEntry zipEntry;
+ final Map<String, String> attributes = new HashMap<>();
while ((zipEntry = zipInputStream.getNextEntry()) != null) {
- processEntry(zipInputStream, zipEntry.isDirectory(), zipEntry.getName(), EncryptionMethod.NONE);
+ addEncryptionMethodAttribute(EncryptionMethod.NONE, attributes);
+ addFileSizeAttribute(zipEntry.getSize(), attributes);
+ addFilePermissionsAttribute(zipEntry.getUnixMode(), attributes);
+ // NOTE: Per javadocs, ZipArchiveEntry can return -1 for getTime() if its not specified
+ // and getLastAccessTime() can return null if it is not specified.
+ Instant lastModified = zipEntry.getLastModifiedDate().toInstant();
+ Instant creation = zipEntry.getTime() > 0 ? new Date(zipEntry.getTime()).toInstant() : null;
+ Instant lastAccess = zipEntry.getLastAccessTime() != null ? zipEntry.getLastAccessTime().toInstant() : null;
+ addZipEntryTimeAttributes(lastModified, creation, lastAccess, attributes);
+ processEntry(zipInputStream, zipEntry.isDirectory(), zipEntry.getName(), attributes);
+ attributes.clear();
}
}
}
@@ -547,8 +616,15 @@ public class UnpackContent extends AbstractProcessor {
public void process(final InputStream inputStream) throws IOException {
try (final ZipInputStream zipInputStream = new ZipInputStream(new BufferedInputStream(inputStream), password, filenameEncoding)) {
LocalFileHeader zipEntry;
+ final Map<String, String> attributes = new HashMap<>();
while ((zipEntry = zipInputStream.getNextEntry()) != null) {
- processEntry(zipInputStream, zipEntry.isDirectory(), zipEntry.getFileName(), zipEntry.getEncryptionMethod());
+ //NOTE: LocalFileHeader has no methods to return creation time and the mode.
+ addEncryptionMethodAttribute(zipEntry.getEncryptionMethod(), attributes);
+ addFileSizeAttribute(zipEntry.getUncompressedSize(), attributes);
+ Instant lastModified = zipEntry.getLastModifiedTime() > 0 ? new Date(zipEntry.getLastModifiedTime()).toInstant() : null;
+ addZipEntryTimeAttributes(lastModified, null, null, attributes);
+ processEntry(zipInputStream, zipEntry.isDirectory(), zipEntry.getFileName(), attributes);
+ attributes.clear();
}
}
}
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 fc2b0226a7..b3207d4494 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
@@ -41,6 +41,7 @@ import java.util.UUID;
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.assertTrue;
@@ -50,6 +51,8 @@ public class TestUnpackContent {
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");
+
@Test
public void testTar() throws IOException {
final TestRunner unpackRunner = TestRunners.newTestRunner(new UnpackContent());
@@ -82,19 +85,21 @@ public class TestUnpackContent {
final List<MockFlowFile> unpacked = unpackRunner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS);
for (final MockFlowFile flowFile : unpacked) {
+ assertTrue(flowFile.getAttributes().keySet().containsAll(List.of(UnpackContent.FRAGMENT_ID, UnpackContent.FRAGMENT_INDEX,
+ UnpackContent.FRAGMENT_COUNT, UnpackContent.SEGMENT_ORIGINAL_FILENAME, UnpackContent.FILE_SIZE_ATTRIBUTE)));
+
final String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
final String folder = flowFile.getAttribute(CoreAttributes.PATH.key());
final Path path = dataPath.resolve(folder).resolve(filename);
- assertEquals("rw-r--r--", flowFile.getAttribute("file.permissions"));
- assertEquals("jmcarey", flowFile.getAttribute("file.owner"));
- assertEquals("mkpasswd", flowFile.getAttribute("file.group"));
- String modifiedTimeAsString = flowFile.getAttribute("file.lastModifiedTime");
- DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ").parse(modifiedTimeAsString);
+ assertEquals("rw-r--r--", flowFile.getAttribute(UnpackContent.FILE_PERMISSIONS_ATTRIBUTE));
+ assertEquals("jmcarey", flowFile.getAttribute(UnpackContent.FILE_OWNER_ATTRIBUTE));
+ assertEquals("mkpasswd", flowFile.getAttribute(UnpackContent.FILE_GROUP_ATTRIBUTE));
+ String modifiedTimeAsString = flowFile.getAttribute("file.lastModifiedTime");
+ assertDoesNotThrow(() -> TIMESTAMP_FORMATTER.parse(modifiedTimeAsString));
String creationTimeAsString = flowFile.getAttribute("file.creationTime");
-
- DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ").parse(creationTimeAsString);
+ assertDoesNotThrow(() -> TIMESTAMP_FORMATTER.parse(creationTimeAsString));
assertTrue(Files.exists(path));
@@ -182,7 +187,24 @@ public class TestUnpackContent {
autoUnpackRunner.assertTransferCount(UnpackContent.REL_FAILURE, 0);
final List<MockFlowFile> unpacked = unpackRunner.getFlowFilesForRelationship(UnpackContent.REL_SUCCESS);
+
+ final List<String> expectedAttributeNames = List.of(
+ CoreAttributes.FILENAME.key(),
+ CoreAttributes.PATH.key(),
+ UnpackContent.FRAGMENT_ID,
+ UnpackContent.FRAGMENT_INDEX,
+ UnpackContent.FRAGMENT_COUNT,
+ UnpackContent.SEGMENT_ORIGINAL_FILENAME,
+ UnpackContent.FILE_SIZE_ATTRIBUTE,
+ UnpackContent.FILE_CREATION_TIME_ATTRIBUTE,
+ UnpackContent.FILE_LAST_MODIFIED_TIME_ATTRIBUTE,
+ UnpackContent.FILE_PERMISSIONS_ATTRIBUTE
+ );
+
for (final MockFlowFile flowFile : unpacked) {
+ for (final String expectedAttributeName : expectedAttributeNames) {
+ flowFile.assertAttributeExists(expectedAttributeName);
+ }
final String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
final String folder = flowFile.getAttribute(CoreAttributes.PATH.key());
final Path path = dataPath.resolve(folder).resolve(filename);
@@ -191,6 +213,7 @@ public class TestUnpackContent {
flowFile.assertContentEquals(path.toFile());
}
}
+
@Test
public void testInvalidZip() throws IOException {
final TestRunner unpackRunner = TestRunners.newTestRunner(new UnpackContent());
@@ -218,7 +241,6 @@ public class TestUnpackContent {
final List<MockFlowFile> unpacked = unpackRunner.getFlowFilesForRelationship(UnpackContent.REL_FAILURE);
for (final MockFlowFile flowFile : unpacked) {
final String filename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
- // final String folder = flowFile.getAttribute(CoreAttributes.PATH.key());
final Path path = dataPath.resolve(filename);
assertTrue(Files.exists(path));