NIFI-14110 Support to limit content size in PackageFlowFile #9595

EndzeitBegins · 2024-12-25T12:21:05Z

Due to using a different API to retrieve the FlowFiles the behaviour when working with multiple queues is no longer unspecified.

I had an circular dependency problem when depending on nifi-mock from nifi-utils, which is why I use an anonymous implementation of FlowFile inside the tests instead of MockFlowFile.

Summary

NIFI-14110

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

Apache NiFi Jira issue created

Pull Request Tracking

Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000

Pull Request Formatting

Pull Request based on current revision of the main branch
Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

Build completed using mvn clean install -P contrib-check
- JDK 21

Licensing

New dependencies are compatible with the Apache License 2.0 according to the License Policy
New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

Documentation formatting appears as expected in rendered files

Due to using a different API to retrieve the FlowFiles the behaviour when working with multiple queues is no longer unspecified.

joewitt · 2024-12-27T16:26:53Z

nifi-commons/nifi-utils/src/main/java/org/apache/nifi/processor/util/FlowFileFilters.java

                    return FlowFileFilterResult.ACCEPT_AND_CONTINUE;
                }

-                if ((size + flowFile.getSize() > maxBytes) || (count + 1 > maxCount)) {
+                if (size > maxBytes || count > maxCount) {


In the previous code count and size were not changed for this case. The case of reject and terminate is terminal and this FlowFileFilterResult will not be further used nor those values in anyway referenced?

Thank you for your review feedback @joewitt.
I'm not sure I correctly understood your comment. Please clarify if the following doesn't address your point.

My goal was to simplify the implementation by unifying the three points of computation (inside the first if statement, inside the predicate of the second if, and before the last return) into a single point.
I added test cases for a few selected scenarios before adjusting the implementation to ensure that the behavior remains the same.

The JavaDoc states:

Returns a new {@link FlowFileFilter} that will pull FlowFiles until the maximum file size has been reached, or the maximum FlowFile Count was been reached ...

Therefore, once either the maximum count or accumulated size limit is reached, all subsequent FlowFiles should be rejected.

Upon closer inspection, the old implementation contained a minor bug related to this behavior. If a FlowFile exceeding the size limit was initially rejected (with REJECT_AND_TERMINATE), and the caller subsequently passed an empty FlowFile to the filter, the filter would incorrectly return ACCEPT_AND_CONTINUE even though the limit had already been reached. While this behavior is technically incorrect, it's primarily a consequence of the caller not adhering to the FlowFileFilterResult contract and thus not a critical bug.
I enhanced one of the test cases to demonstrate this behavior.

Both the old and new implementations are susceptible to potential integer overflow issues for both count and size. I'm not sure if that's something worth addressing though.

I'm content with leaving the implementation of newSizeBasedFilter unchanged if that's prefered. I encountered the code while working on the desired changes for PackageFlowFile and deemed it worth some tests / minor refactor.

I dove in to understand this code and the framework's usage of it, and the changes look good to me, and are a bit more readable. The framework does not currently call filter() again if the FilterResult indicates TERMINATE. But if this changes in the future, this code continues to return REJECT_AND_TERMINATE on following calls to filter(), as expected.

…imit was reached

mosermw · 2024-12-30T20:15:40Z

I recommend modifying a MultiProcessorUseCase or creating another UseCase in order to make the documentation for combining the two Batch Size properties clear. It's very important to be clear that flowfiles will not be delayed in the input queue waiting for a batch size to be reached. It's also very important to support packaging exactly 1 flowfile.

While this improvement appears to be worthwhile, we should be very careful with configuration creep on PackageFlowFile. It's only justification for existence is to be easier to use than MergeContent for a specific use case. Too many features would ruin that justification.

EndzeitBegins · 2025-01-04T14:49:06Z

Thank you for the useful feedback @mosermw. I've adjusted the documentation of the UseCases to clarify the batching behaviour of the processor.

PackageFlowFile in combination with UnpackContent is a useful pair of processors to transfer FlowFiles between NiFi clusters where the more robust approach using remote process groups is not applicable, e.g. due to network restrictions.
Packaging more than one FlowFile can improve efficiency both in storage and transmission.
In my opinion, when the content size of the FlowFiles to transfer can vary largely, being able to apply a soft constraint on the package size can be helpful.

Personally I do not intent to add other properties to the processor at the moment.

mosermw

I reviewed and tested this in various scenarios in running NiFi and just have a few minor comments.

mosermw · 2025-03-17T20:09:42Z

nifi-commons/nifi-utils/src/main/java/org/apache/nifi/processor/util/FlowFileFilters.java

                    return FlowFileFilterResult.ACCEPT_AND_CONTINUE;
                }

-                if ((size + flowFile.getSize() > maxBytes) || (count + 1 > maxCount)) {
+                if (size > maxBytes || count > maxCount) {


I dove in to understand this code and the framework's usage of it, and the changes look good to me, and are a bit more readable. The framework does not currently call filter() again if the FilterResult indicates TERMINATE. But if this changes in the future, this code continues to return REJECT_AND_TERMINATE on following calls to filter(), as expected.

mosermw · 2025-03-17T20:10:05Z

nifi-commons/nifi-utils/src/test/java/org/apache/nifi/processor/util/FlowFileFiltersTest.java

+    private int claimFlowFileId() {
+        return flowFileId++;
+    }
+}


Would you add a newline to the end of this file?

mosermw · 2025-03-17T20:10:34Z

...i-standard-processors/src/main/java/org/apache/nifi/processors/standard/PackageFlowFile.java


+                    It can be more efficient to transmit one larger package of batched FlowFiles than each FlowFile packaged separately.


This line appears to be redundant with the BATCHING_BEHAVIOUR_DESCRIPTION, should we remove it?

mosermw · 2025-03-17T20:11:54Z

...i-standard-processors/src/main/java/org/apache/nifi/processors/standard/PackageFlowFile.java

+            .description("Maximum combined content size of FlowFiles to package into one output FlowFile. " +
+                    "Note, that FlowFiles whose content exceeds this limit are packaged separately.")
+            .required(true)
+            .defaultValue("1 GB")


The default Maximum Batch Size is "1", indicating a desire for default configuration to package 1 FlowFile per output. I think we should maintain that default behavior with this new Maximum Batch Content Size property. Would you make its default something like "0 MB" so that 1 FlowFile is packaged?

mosermw · 2025-03-17T20:12:52Z

...i-standard-processors/src/main/java/org/apache/nifi/processors/standard/PackageFlowFile.java

+                "Maximum Batch Size" > 1 can improve storage or transmission efficiency by batching many FlowFiles together into 1 larger file.
+                "Maximum Batch Content Size" can be used to enforce a soft upper limit on the overall package size.
+
+                Note, that the Batch properties only restrict the maximum amount of FlowFiles to incorporate into a single package.


Thanks for including this note! It's important.

mosermw · 2025-03-17T20:13:50Z

...i-standard-processors/src/main/java/org/apache/nifi/processors/standard/PackageFlowFile.java

+            .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+            .build();
+
+    private static final List<PropertyDescriptor> PROPERTIES = List.of(BATCH_SIZE, BATCH_CONTENT_SIZE);


There is a conflict with the main branch related to the List of PropertyDescriptors. By convention, try to keep 1 PropertyDescriptor per line for readability.

NIFI-14110 Support to limit content size in PackageFlowFile

3715edb

Due to using a different API to retrieve the FlowFiles the behaviour when working with multiple queues is no longer unspecified.

joewitt reviewed Dec 27, 2024

View reviewed changes

NIFI-14110 Enhance tests to ensure FlowFiles are rejected once size l…

4da4d0b

…imit was reached

NIFI-14110 Explain batching behaviour in UseCases

908427e

mosermw requested changes Mar 17, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

NIFI-14110 Support to limit content size in PackageFlowFile #9595

NIFI-14110 Support to limit content size in PackageFlowFile #9595

EndzeitBegins commented Dec 25, 2024 •

edited

Loading

joewitt Dec 27, 2024

EndzeitBegins Dec 28, 2024 •

edited

Loading

mosermw Mar 17, 2025

mosermw commented Dec 30, 2024

EndzeitBegins commented Jan 4, 2025

mosermw left a comment

mosermw Mar 17, 2025

mosermw Mar 17, 2025

mosermw Mar 17, 2025

mosermw Mar 17, 2025

mosermw Mar 17, 2025

mosermw Mar 17, 2025


		It can be more efficient to transmit one larger package of batched FlowFiles than each FlowFile packaged separately.

NIFI-14110 Support to limit content size in PackageFlowFile #9595

Are you sure you want to change the base?

NIFI-14110 Support to limit content size in PackageFlowFile #9595

Conversation

EndzeitBegins commented Dec 25, 2024 • edited Loading

Summary

Tracking

Issue Tracking

Pull Request Tracking

Pull Request Formatting

Verification

Build

Licensing

Documentation

Choose a reason for hiding this comment

EndzeitBegins Dec 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mosermw commented Dec 30, 2024

EndzeitBegins commented Jan 4, 2025

mosermw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

EndzeitBegins commented Dec 25, 2024 •

edited

Loading

EndzeitBegins Dec 28, 2024 •

edited

Loading