You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nifi.apache.org by rickysaltzer <gi...@git.apache.org> on 2016/02/25 23:25:12 UTC
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
GitHub user rickysaltzer opened a pull request:
https://github.com/apache/nifi/pull/248
NIFI-1568: Add Filter Capability to UnpackContent
Adds a "File Filter" property to the `UnpackContent` processor
to allow users to specify which files are eligible for extraction.
By default, all files will be extracted.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/rickysaltzer/nifi NIFI-1568
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/248.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #248
----
commit 89d2a8565284e985411027db37dff69f9faf3592
Author: ricky <ri...@cloudera.com>
Date: 2016-02-25T22:20:51Z
NIFI-1568: Add Filter Capability to UnpackContent
Adds a "File Filter" property to the `UnpackContent` processor
to allow users to specify which files are eligible for extraction.
By default, all files will be extracted.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by trkurc <gi...@git.apache.org>.
Github user trkurc commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r64084293
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -111,8 +107,18 @@
.name("Packaging Format")
.description("The Packaging Format used to create the file")
.required(true)
- .allowableValues(AUTO_DETECT_FORMAT, TAR_FORMAT, ZIP_FORMAT, FLOWFILE_STREAM_FORMAT_V3, FLOWFILE_STREAM_FORMAT_V2, FLOWFILE_TAR_FORMAT)
- .defaultValue(AUTO_DETECT_FORMAT)
+ .allowableValues(PackageFormat.AUTO_DETECT_FORMAT.toString(), PackageFormat.TAR_FORMAT.toString(),
+ PackageFormat.ZIP_FORMAT.toString(), PackageFormat.FLOWFILE_STREAM_FORMAT_V3.toString(),
+ PackageFormat.FLOWFILE_STREAM_FORMAT_V2.toString(), PackageFormat.FLOWFILE_TAR_FORMAT.toString())
+ .defaultValue(PackageFormat.AUTO_DETECT_FORMAT.toString())
+ .build();
+
+ public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+ .name("File Filter")
+ .description("Only files whose names match the given regular expression will be extracted (tar/zip only)")
+ .required(true)
+ .defaultValue("[^\\.].*")
--- End diff --
@rickysaltzer - I'm a bit confused by this default value... is the default to filter files starting with "."? Does this change this processor's behavior?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r67017365
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,88 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ComponentLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
case TAR_FORMAT:
- unpacker = new TarUnpacker();
+ case X_TAR_FORMAT:
+ unpacker = tarUnpacker;
addFragmentAttrs = true;
break;
case ZIP_FORMAT:
- unpacker = new ZipUnpacker();
+ unpacker = zipUnpacker;
addFragmentAttrs = true;
break;
case FLOWFILE_STREAM_FORMAT_V2:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ unpacker = flowFileStreamV2Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_STREAM_FORMAT_V3:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ unpacker = flowFileStreamV3Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_TAR_FORMAT:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ unpacker = flowFileTarUnpacker;
addFragmentAttrs = false;
break;
- default:
- throw new AssertionError("Packaging Format was " + context.getProperty(PACKAGING_FORMAT).getValue());
+ case AUTO_DETECT_FORMAT:
+ // The format of the unpacker should be known before initialization
+ throw new ProcessException(packagingFormat + " is not a valid packaging format");
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog logger = getLogger();
+ PackageFormat packagingFormat = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase());
+ if (packagingFormat == PackageFormat.AUTO_DETECT_FORMAT) {
+ packagingFormat = null;
+ final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
+ if (mimeType == null) {
+ logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ for (PackageFormat format: PackageFormat.values()) {
+ if (mimeType.toLowerCase().equals(format.getMimeType())) {
+ packagingFormat = format;
+ }
+ }
+ if (packagingFormat == null) {
+ logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
Should this be routed to something like a "not unpacked" relationship, rather than success? Seems like outgoing flow files may or may not be unpacked, and you'd have to do some additional logic (like check mime type or something) to see if it worked.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on the pull request:
https://github.com/apache/nifi/pull/248#issuecomment-211408619
sure thing, @olegz. I just got back from vacation today, so I will try and take a look this week. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r67604557
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,88 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ComponentLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
case TAR_FORMAT:
- unpacker = new TarUnpacker();
+ case X_TAR_FORMAT:
+ unpacker = tarUnpacker;
addFragmentAttrs = true;
break;
case ZIP_FORMAT:
- unpacker = new ZipUnpacker();
+ unpacker = zipUnpacker;
addFragmentAttrs = true;
break;
case FLOWFILE_STREAM_FORMAT_V2:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ unpacker = flowFileStreamV2Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_STREAM_FORMAT_V3:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ unpacker = flowFileStreamV3Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_TAR_FORMAT:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ unpacker = flowFileTarUnpacker;
addFragmentAttrs = false;
break;
- default:
- throw new AssertionError("Packaging Format was " + context.getProperty(PACKAGING_FORMAT).getValue());
+ case AUTO_DETECT_FORMAT:
+ // The format of the unpacker should be known before initialization
+ throw new ProcessException(packagingFormat + " is not a valid packaging format");
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog logger = getLogger();
+ PackageFormat packagingFormat = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase());
+ if (packagingFormat == PackageFormat.AUTO_DETECT_FORMAT) {
+ packagingFormat = null;
+ final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
+ if (mimeType == null) {
+ logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ for (PackageFormat format: PackageFormat.values()) {
+ if (mimeType.toLowerCase().equals(format.getMimeType())) {
+ packagingFormat = format;
+ }
+ }
+ if (packagingFormat == null) {
+ logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
Great question, @mattyb149. That is pretty confusing to be honest. It looks like the old code also does this. I could implement your suggestion, but the only downside is any pre-existing flows will have to be updated because each of the processors will be in an invalid state. If this is acceptable, I'll go ahead and throw it in. Thoughts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/248
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r54299720
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -202,11 +213,11 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
final boolean addFragmentAttrs;
switch (packagingFormat) {
case TAR_FORMAT:
--- End diff --
I would. And if there are ways to improve test coverage that would help eliminate the "fear it would break" in the future ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on the pull request:
https://github.com/apache/nifi/pull/248#issuecomment-218596040
@olegz sorry for the delay on this, I went ahead and added a case label for the `AUTO_DETECT_FORMAT` enum.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on the issue:
https://github.com/apache/nifi/pull/248
Looks like the travis-ci build went through and there's no conflicts with the base branch :). Let me know if you have any questions about testing out the patch. Thanks again!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r54298517
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -202,11 +213,11 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
final boolean addFragmentAttrs;
switch (packagingFormat) {
case TAR_FORMAT:
--- End diff --
yeah I noticed the same thing when I added the filter ability. I didn't want to make any major refactoring to it in fear it would break. I could give it a try though, because I agree, it doesn't make sense to have this logic for each incoming flowfile.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on the pull request:
https://github.com/apache/nifi/pull/248#issuecomment-220686739
@trkurc changed the default value to `.*`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r59990453
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,85 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ProcessorLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
--- End diff --
This one complains with the compile-warning "The enum constant AUTO_DETECT_FORMAT needs a corresponding case label in this enum switch on UnpackContent.PackageFormat". Can you take a look?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r67605391
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,88 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ComponentLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
case TAR_FORMAT:
- unpacker = new TarUnpacker();
+ case X_TAR_FORMAT:
+ unpacker = tarUnpacker;
addFragmentAttrs = true;
break;
case ZIP_FORMAT:
- unpacker = new ZipUnpacker();
+ unpacker = zipUnpacker;
addFragmentAttrs = true;
break;
case FLOWFILE_STREAM_FORMAT_V2:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ unpacker = flowFileStreamV2Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_STREAM_FORMAT_V3:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ unpacker = flowFileStreamV3Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_TAR_FORMAT:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ unpacker = flowFileTarUnpacker;
addFragmentAttrs = false;
break;
- default:
- throw new AssertionError("Packaging Format was " + context.getProperty(PACKAGING_FORMAT).getValue());
+ case AUTO_DETECT_FORMAT:
+ // The format of the unpacker should be known before initialization
+ throw new ProcessException(packagingFormat + " is not a valid packaging format");
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog logger = getLogger();
+ PackageFormat packagingFormat = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase());
+ if (packagingFormat == PackageFormat.AUTO_DETECT_FORMAT) {
+ packagingFormat = null;
+ final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
+ if (mimeType == null) {
+ logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ for (PackageFormat format: PackageFormat.values()) {
+ if (mimeType.toLowerCase().equals(format.getMimeType())) {
+ packagingFormat = format;
+ }
+ }
+ if (packagingFormat == null) {
+ logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
I'm fine with keeping the current behavior, just wanted to discuss first to make sure I understood :) thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by joewitt <gi...@git.apache.org>.
Github user joewitt commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r67604647
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,88 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ComponentLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
case TAR_FORMAT:
- unpacker = new TarUnpacker();
+ case X_TAR_FORMAT:
+ unpacker = tarUnpacker;
addFragmentAttrs = true;
break;
case ZIP_FORMAT:
- unpacker = new ZipUnpacker();
+ unpacker = zipUnpacker;
addFragmentAttrs = true;
break;
case FLOWFILE_STREAM_FORMAT_V2:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ unpacker = flowFileStreamV2Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_STREAM_FORMAT_V3:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ unpacker = flowFileStreamV3Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_TAR_FORMAT:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ unpacker = flowFileTarUnpacker;
addFragmentAttrs = false;
break;
- default:
- throw new AssertionError("Packaging Format was " + context.getProperty(PACKAGING_FORMAT).getValue());
+ case AUTO_DETECT_FORMAT:
+ // The format of the unpacker should be known before initialization
+ throw new ProcessException(packagingFormat + " is not a valid packaging format");
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog logger = getLogger();
+ PackageFormat packagingFormat = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase());
+ if (packagingFormat == PackageFormat.AUTO_DETECT_FORMAT) {
+ packagingFormat = null;
+ final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
+ if (mimeType == null) {
+ logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ for (PackageFormat format: PackageFormat.values()) {
+ if (mimeType.toLowerCase().equals(format.getMimeType())) {
+ packagingFormat = format;
+ }
+ }
+ if (packagingFormat == null) {
+ logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
I agree the naming of 'success' is unfortunate given this case. That said not sure it would be worth the 'weight' so to speak of adding another relationship just for this case because it should be pretty rare/odd that something would go that route. I would be fine if you guys do decide to go that route (even though yes it means flows would need adjustment) just saying it may not really be worth it. We do allow for this sort of change in our versioning guidance along a minor release. Making it so that a flow becomes invalid is ok if the juice is considered worth the squeeze. What is not ok is fundamentally changing the behavior of a processor such that flows could be broken with the user having a chance to know that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by rickysaltzer <gi...@git.apache.org>.
Github user rickysaltzer commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r64087730
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -111,8 +107,18 @@
.name("Packaging Format")
.description("The Packaging Format used to create the file")
.required(true)
- .allowableValues(AUTO_DETECT_FORMAT, TAR_FORMAT, ZIP_FORMAT, FLOWFILE_STREAM_FORMAT_V3, FLOWFILE_STREAM_FORMAT_V2, FLOWFILE_TAR_FORMAT)
- .defaultValue(AUTO_DETECT_FORMAT)
+ .allowableValues(PackageFormat.AUTO_DETECT_FORMAT.toString(), PackageFormat.TAR_FORMAT.toString(),
+ PackageFormat.ZIP_FORMAT.toString(), PackageFormat.FLOWFILE_STREAM_FORMAT_V3.toString(),
+ PackageFormat.FLOWFILE_STREAM_FORMAT_V2.toString(), PackageFormat.FLOWFILE_TAR_FORMAT.toString())
+ .defaultValue(PackageFormat.AUTO_DETECT_FORMAT.toString())
+ .build();
+
+ public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+ .name("File Filter")
+ .description("Only files whose names match the given regular expression will be extracted (tar/zip only)")
+ .required(true)
+ .defaultValue("[^\\.].*")
--- End diff --
@trkurc - I believe it does, I took it from the `GetFile` processor, but it doesn't really make sense here. I think changing this to just `.*` would make more sense. I'll do that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r67246009
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -154,75 +171,88 @@ protected void init(final ProcessorInitializationContext context) {
return properties;
}
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
+ @OnStopped
+ public void onStopped() {
+ unpacker = null;
+ fileFilter = null;
+ }
- final ComponentLog logger = getLogger();
- String packagingFormat = context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase();
- if (AUTO_DETECT_FORMAT.equals(packagingFormat)) {
- final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
- if (mimeType == null) {
- logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
+ @OnScheduled
+ public void onScheduled(ProcessContext context) throws ProcessException {
+ if (fileFilter == null) {
+ fileFilter = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+ tarUnpacker = new TarUnpacker(fileFilter);
+ zipUnpacker = new ZipUnpacker(fileFilter);
+ flowFileStreamV3Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ flowFileStreamV2Unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ flowFileTarUnpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ }
- switch (mimeType.toLowerCase()) {
- case "application/tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/x-tar":
- packagingFormat = TAR_FORMAT;
- break;
- case "application/zip":
- packagingFormat = ZIP_FORMAT;
- break;
- case "application/flowfile-v3":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V3;
- break;
- case "application/flowfile-v2":
- packagingFormat = FLOWFILE_STREAM_FORMAT_V2;
- break;
- case "application/flowfile-v1":
- packagingFormat = FLOWFILE_TAR_FORMAT;
- break;
- default: {
- logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
- session.transfer(flowFile, REL_SUCCESS);
- return;
- }
- }
+ PackageFormat format = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue());
+ if (format != PackageFormat.AUTO_DETECT_FORMAT && unpacker == null) {
+ initUnpacker(format);
}
+ }
- final Unpacker unpacker;
- final boolean addFragmentAttrs;
+ public void initUnpacker(PackageFormat packagingFormat) {
switch (packagingFormat) {
case TAR_FORMAT:
- unpacker = new TarUnpacker();
+ case X_TAR_FORMAT:
+ unpacker = tarUnpacker;
addFragmentAttrs = true;
break;
case ZIP_FORMAT:
- unpacker = new ZipUnpacker();
+ unpacker = zipUnpacker;
addFragmentAttrs = true;
break;
case FLOWFILE_STREAM_FORMAT_V2:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV2());
+ unpacker = flowFileStreamV2Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_STREAM_FORMAT_V3:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV3());
+ unpacker = flowFileStreamV3Unpacker;
addFragmentAttrs = false;
break;
case FLOWFILE_TAR_FORMAT:
- unpacker = new FlowFileStreamUnpacker(new FlowFileUnpackagerV1());
+ unpacker = flowFileTarUnpacker;
addFragmentAttrs = false;
break;
- default:
- throw new AssertionError("Packaging Format was " + context.getProperty(PACKAGING_FORMAT).getValue());
+ case AUTO_DETECT_FORMAT:
+ // The format of the unpacker should be known before initialization
+ throw new ProcessException(packagingFormat + " is not a valid packaging format");
+ }
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ final ComponentLog logger = getLogger();
+ PackageFormat packagingFormat = PackageFormat.getFormat(context.getProperty(PACKAGING_FORMAT).getValue().toLowerCase());
+ if (packagingFormat == PackageFormat.AUTO_DETECT_FORMAT) {
+ packagingFormat = null;
+ final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
+ if (mimeType == null) {
+ logger.error("No mime.type attribute set for {}; routing to failure", new Object[]{flowFile});
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ for (PackageFormat format: PackageFormat.values()) {
+ if (mimeType.toLowerCase().equals(format.getMimeType())) {
+ packagingFormat = format;
+ }
+ }
+ if (packagingFormat == null) {
+ logger.info("Cannot unpack {} because its mime.type attribute is set to '{}', which is not a format that can be unpacked; routing to 'success'", new Object[]{flowFile, mimeType});
+ session.transfer(flowFile, REL_SUCCESS);
--- End diff --
@rickysaltzer everything LGTM but wante to get your opinion on the above, will merge once addressed/discussed :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #248: NIFI-1568: Add Filter Capability to UnpackContent
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/248
I can continue the review on this one, mind rebasing against the latest master or 0.x branch? Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request: NIFI-1568: Add Filter Capability to UnpackConte...
Posted by olegz <gi...@git.apache.org>.
Github user olegz commented on a diff in the pull request:
https://github.com/apache/nifi/pull/248#discussion_r54298253
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/UnpackContent.java ---
@@ -202,11 +213,11 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
final boolean addFragmentAttrs;
switch (packagingFormat) {
case TAR_FORMAT:
--- End diff --
Have you considered doing some restructuring of the _onTrigger()_ method?
Since the packaging format is based on the processor wide configuration property (not FlowFile) most of the logic (switch/case) could be performed once instead of every time _onTrigger()_ is called. That includes the new logic with regex pattern compile.
Or am I missing something big here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---