You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2025/01/16 17:10:38 UTC
(nifi) branch main updated: NIFI-14166: Added "Bin Termination Check" property to MergeContent; updated documentation to include the new property and more clearly explain when bins get merged.
This is an automated email from the ASF dual-hosted git repository.
pvillard 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 2115d68ae7 NIFI-14166: Added "Bin Termination Check" property to MergeContent; updated documentation to include the new property and more clearly explain when bins get merged.
2115d68ae7 is described below
commit 2115d68ae7729857d823cb49f8180c8a9ce4302e
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Thu Jan 16 09:19:29 2025 -0500
NIFI-14166: Added "Bin Termination Check" property to MergeContent; updated documentation to include the new property and more clearly explain when bins get merged.
Signed-off-by: Pierre Villard <pi...@gmail.com>
This closes #9639.
---
.../org/apache/nifi/processor/util/bin/Bin.java | 35 +++--
.../apache/nifi/processor/util/bin/BinManager.java | 114 ++++++++--------
.../nifi/processor/util/bin/EvictionReason.java | 2 +
...{EvictionReason.java => InsertionLocation.java} | 36 +++---
.../nifi/processors/standard/MergeContent.java | 38 ++++++
.../additionalDetails.md | 22 +++-
.../nifi/processors/standard/TestMergeContent.java | 143 +++++++++++++++++++++
7 files changed, 303 insertions(+), 87 deletions(-)
diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/Bin.java b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/Bin.java
index 30797de200..a00c5adac7 100644
--- a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/Bin.java
+++ b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/Bin.java
@@ -48,6 +48,7 @@ public class Bin {
private volatile int maximumEntries;
private final String fileCountAttribute;
private volatile EvictionReason evictionReason = EvictionReason.UNSET;
+ private volatile boolean forcefullyCompleted = false;
private final List<FlowFile> binContents = new ArrayList<>();
private final Set<String> binIndexSet = new HashSet<>();
@@ -93,6 +94,14 @@ public class Bin {
return session;
}
+ public void complete() {
+ forcefullyCompleted = true;
+ }
+
+ public boolean isForcefullyCompleted() {
+ return forcefullyCompleted;
+ }
+
/**
* Indicates whether the bin has enough items to be considered full. This is based on whether the current size of the bin is greater than the minimum size in bytes and based on having a number of
* successive unsuccessful attempts to add a new item (because it is so close to the max or the size of the objects being attempted do not favor tight packing)
@@ -101,10 +110,14 @@ public class Bin {
*/
public boolean isFull() {
return (((size >= minimumSizeBytes) && binContents.size() >= minimumEntries) && (successiveFailedOfferings > 5))
- || (size >= maximumSizeBytes) || (binContents.size() >= maximumEntries);
+ || (size >= maximumSizeBytes) || (binContents.size() >= maximumEntries)
+ || forcefullyCompleted;
}
- public EvictionReason determineFullness() {
+ public EvictionReason determineEvictionReason() {
+ if (forcefullyCompleted) {
+ return EvictionReason.BIN_TERMINATION_SIGNAL;
+ }
if (size >= maximumSizeBytes) {
return EvictionReason.MAX_BYTES_THRESHOLD_REACHED;
}
@@ -165,21 +178,13 @@ public class Bin {
* @return true if added; false otherwise
*/
public boolean offer(final FlowFile flowFile, final ProcessSession session) {
- if (((size + flowFile.getSize()) > maximumSizeBytes) || (binContents.size() >= maximumEntries)) {
+ if (forcefullyCompleted || ((size + flowFile.getSize()) > maximumSizeBytes) || (binContents.size() >= maximumEntries)) {
successiveFailedOfferings++;
return false;
}
// fileCountAttribute is non-null for defragment mode
if (fileCountAttribute != null) {
- final String countValue = flowFile.getAttribute(fileCountAttribute);
- final Integer count = toInteger(countValue);
- if (count != null) {
- // set the limits for the bin as an exact count when the count attribute arrives
- this.maximumEntries = count;
- this.minimumEntries = count;
- }
-
final String index = flowFile.getAttribute(FRAGMENT_INDEX_ATTRIBUTE);
if (index == null || index.isEmpty() || !binIndexSet.add(index)) {
// Do not accept flowfile with duplicate fragment index value
@@ -187,6 +192,14 @@ public class Bin {
successiveFailedOfferings++;
return false;
}
+
+ final String countValue = flowFile.getAttribute(fileCountAttribute);
+ final Integer count = toInteger(countValue);
+ if (count != null) {
+ // set the limits for the bin as an exact count when the count attribute arrives
+ this.maximumEntries = count;
+ this.minimumEntries = count;
+ }
}
size += flowFile.getSize();
diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/BinManager.java b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/BinManager.java
index 72ce099b1f..5e611d29fa 100644
--- a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/BinManager.java
+++ b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/BinManager.java
@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Predicate;
/**
* This class is thread safe
@@ -45,6 +46,8 @@ public class BinManager {
private final AtomicInteger minEntries = new AtomicInteger(0);
private final AtomicInteger maxEntries = new AtomicInteger(Integer.MAX_VALUE);
private final AtomicReference<String> fileCountAttribute = new AtomicReference<>(null);
+ private volatile Predicate<FlowFile> binTerminationCheck = ff -> false;
+ private volatile InsertionLocation insertionLocation = InsertionLocation.LAST_IN_BIN;
private final AtomicInteger maxBinAgeSeconds = new AtomicInteger(Integer.MAX_VALUE);
private final Map<String, List<Bin>> groupBinMap = new HashMap<>();
@@ -88,6 +91,16 @@ public class BinManager {
this.maxEntries.set(maximumEntries);
}
+ /**
+ * Sets the predicate that determines whether or not a FlowFile should terminate a bin and the location in the bin where the FlowFile should be inserted if so.
+ * @param binTerminationCheck the predicate to use to determine if a FlowFile should terminate a bin
+ * @param insertionLocation the location in the bin where the FlowFile should be inserted if it terminates the bin
+ */
+ public void setBinTermination(final Predicate<FlowFile> binTerminationCheck, final InsertionLocation insertionLocation) {
+ this.binTerminationCheck = binTerminationCheck;
+ this.insertionLocation = insertionLocation;
+ }
+
public int getBinCount() {
rLock.lock();
try {
@@ -121,39 +134,8 @@ public class BinManager {
* @return true if added; false if no bin exists which can fit this item and no bin can be created based on current min/max criteria
*/
public boolean offer(final String groupIdentifier, final FlowFile flowFile, final ProcessSession session, final ProcessSessionFactory sessionFactory) {
- final long currentMaxSizeBytes = maxSizeBytes.get();
- if (flowFile.getSize() > currentMaxSizeBytes) { //won't fit into any new bins (and probably none existing)
- return false;
- }
- wLock.lock();
- try {
- final List<Bin> currentBins = groupBinMap.get(groupIdentifier);
- if (currentBins == null) { // this is a new group we need to register
- final List<Bin> bins = new ArrayList<>();
- final Bin bin = new Bin(sessionFactory.createSession(), minSizeBytes.get(), currentMaxSizeBytes, minEntries.get(),
- maxEntries.get(), fileCountAttribute.get());
- bins.add(bin);
- groupBinMap.put(groupIdentifier, bins);
- binCount++;
- return bin.offer(flowFile, session);
- } else {
- for (final Bin bin : currentBins) {
- final boolean accepted = bin.offer(flowFile, session);
- if (accepted) {
- return true;
- }
- }
-
- //if we've reached this point then we couldn't fit it into any existing bins - gotta make a new one
- final Bin bin = new Bin(sessionFactory.createSession(), minSizeBytes.get(), currentMaxSizeBytes, minEntries.get(),
- maxEntries.get(), fileCountAttribute.get());
- currentBins.add(bin);
- binCount++;
- return bin.offer(flowFile, session);
- }
- } finally {
- wLock.unlock();
- }
+ final Set<FlowFile> unbinned = offer(groupIdentifier, List.of(flowFile), session, sessionFactory);
+ return unbinned.isEmpty();
}
/**
@@ -179,11 +161,39 @@ public class BinManager {
continue;
}
+ final boolean terminatesBin = binTerminationCheck != null && binTerminationCheck.test(flowFile);
+
final List<Bin> currentBins = groupBinMap.computeIfAbsent(groupIdentifier, k -> new ArrayList<>());
- for (final Bin bin : currentBins) {
- final boolean accepted = bin.offer(flowFile, session);
- if (accepted) {
- continue flowFileLoop;
+ if (terminatesBin) {
+ if (insertionLocation == InsertionLocation.LAST_IN_BIN) {
+ for (final Bin bin : currentBins) {
+ final boolean accepted = bin.offer(flowFile, session);
+ if (accepted) {
+ bin.complete();
+ bin.setEvictionReason(EvictionReason.BIN_TERMINATION_SIGNAL);
+
+ continue flowFileLoop;
+ }
+ }
+ } else if (!currentBins.isEmpty()) {
+ for (final Bin bin : currentBins) {
+ if (bin.isForcefullyCompleted()) {
+ continue;
+ }
+
+ bin.complete();
+ bin.setEvictionReason(EvictionReason.BIN_TERMINATION_SIGNAL);
+ break;
+
+ // Note that we intentionally do not continue the flowFileLoop here because we want to create a new bin
+ }
+ }
+ } else {
+ for (final Bin bin : currentBins) {
+ final boolean accepted = bin.offer(flowFile, session);
+ if (accepted) {
+ continue flowFileLoop;
+ }
}
}
@@ -194,7 +204,12 @@ public class BinManager {
currentBins.add(bin);
binCount++;
final boolean added = bin.offer(flowFile, session);
- if (!added) {
+ if (added) {
+ if (terminatesBin && (insertionLocation == InsertionLocation.ISOLATED || insertionLocation == InsertionLocation.LAST_IN_BIN)) {
+ bin.complete();
+ bin.setEvictionReason(EvictionReason.BIN_TERMINATION_SIGNAL);
+ }
+ } else {
unbinned.add(flowFile);
}
@@ -223,10 +238,10 @@ public class BinManager {
final List<Bin> remainingBins = new ArrayList<>();
for (final Bin bin : group.getValue()) {
if (relaxFullnessConstraint && bin.isFullEnough()) {
- bin.setEvictionReason(bin.determineFullness());
+ bin.setEvictionReason(bin.determineEvictionReason());
readyBins.add(bin);
} else if (!relaxFullnessConstraint && bin.isFull()) { //strict check
- bin.setEvictionReason(bin.determineFullness());
+ bin.setEvictionReason(bin.determineEvictionReason());
readyBins.add(bin);
} else if (relaxFullnessConstraint && bin.isOlderThan(maxBinAgeSeconds.get(), TimeUnit.SECONDS)) {
bin.setEvictionReason(EvictionReason.TIMEOUT);
@@ -279,23 +294,4 @@ public class BinManager {
}
}
- /**
- * @return true if any current bins are older than the allowable max
- */
- public boolean containsOldBins() {
- rLock.lock();
- try {
- for (final List<Bin> bins : groupBinMap.values()) {
- for (final Bin bin : bins) {
- if (bin.isOlderThan(maxBinAgeSeconds.get(), TimeUnit.SECONDS)) {
- return true;
- }
- }
- }
- } finally {
- rLock.unlock();
- }
- return false;
- }
-
}
diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java
index ddafc20e66..e113891800 100644
--- a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java
+++ b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java
@@ -28,6 +28,8 @@ public enum EvictionReason {
BIN_MANAGER_FULL("The oldest Bin was removed because incoming FlowFile could not be placed in an existing Bin, and the Maximum Number of Bins was reached"),
+ BIN_TERMINATION_SIGNAL("A FlowFile signaled that the Bin should be terminated"),
+
UNSET("No reason was determined");
private final String explanation;
diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/InsertionLocation.java
similarity index 50%
copy from nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java
copy to nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/InsertionLocation.java
index ddafc20e66..b820c08911 100644
--- a/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/EvictionReason.java
+++ b/nifi-extension-bundles/nifi-extension-utils/nifi-bin-manager/src/main/java/org/apache/nifi/processor/util/bin/InsertionLocation.java
@@ -17,29 +17,33 @@
package org.apache.nifi.processor.util.bin;
-public enum EvictionReason {
- MAX_BYTES_THRESHOLD_REACHED("Maximum number of bytes (Max Group Size) reached"),
+import org.apache.nifi.components.DescribedValue;
- MAX_ENTRIES_THRESHOLD_REACHED("Maximum number of entries reached"),
+public enum InsertionLocation implements DescribedValue {
+ LAST_IN_BIN("Last in Bin", "Insert the FlowFile at the end of the Bin that is terminated"),
+ FIRST_IN_NEW_BIN("First in New Bin", "Insert the FlowFile at the beginning of a newly created Bin"),
+ ISOLATED("Isolated", "Insert the FlowFile into a new Bin and terminate the Bin immediately with the FlowFile as the only content");
- MIN_THRESHOLDS_REACHED("Minimum number of bytes (Min Group Size) and minimum number of entries reached"),
+ private final String name;
+ private final String description;
- TIMEOUT("Max Bin Age reached"),
-
- BIN_MANAGER_FULL("The oldest Bin was removed because incoming FlowFile could not be placed in an existing Bin, and the Maximum Number of Bins was reached"),
-
- UNSET("No reason was determined");
+ InsertionLocation(final String name, final String description) {
+ this.name = name;
+ this.description = description;
+ }
- private final String explanation;
- EvictionReason(final String explanation) {
- this.explanation = explanation;
+ @Override
+ public String getValue() {
+ return name;
}
- public String getExplanation() {
- return explanation;
+ @Override
+ public String getDisplayName() {
+ return name;
}
- public String toString() {
- return explanation;
+ @Override
+ public String getDescription() {
+ return description;
}
}
diff --git a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
index c57576174f..34e25bbbf1 100644
--- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
+++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
@@ -44,10 +44,12 @@ import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.documentation.UseCase;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.resource.ResourceCardinality;
import org.apache.nifi.components.resource.ResourceType;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
@@ -62,6 +64,7 @@ import org.apache.nifi.processor.util.bin.Bin;
import org.apache.nifi.processor.util.bin.BinFiles;
import org.apache.nifi.processor.util.bin.BinManager;
import org.apache.nifi.processor.util.bin.BinProcessingResult;
+import org.apache.nifi.processor.util.bin.InsertionLocation;
import org.apache.nifi.processors.standard.merge.AttributeStrategy;
import org.apache.nifi.processors.standard.merge.AttributeStrategyUtil;
import org.apache.nifi.stream.io.NonCloseableOutputStream;
@@ -95,6 +98,7 @@ import java.util.OptionalLong;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
import java.util.regex.Pattern;
import java.util.zip.ZipEntry;
import java.util.zip.ZipException;
@@ -395,6 +399,31 @@ public class MergeContent extends BinFiles {
.dependsOn(MERGE_FORMAT, MERGE_FORMAT_TAR)
.build();
+ public static final PropertyDescriptor BIN_TERMINATION_CHECK = new PropertyDescriptor.Builder()
+ .name("Bin Termination Check")
+ .description("""
+ Specifies an Expression Language Expression that is to be evaluated against each FlowFile. If the result of the expression is 'true', the
+ bin that the FlowFile corresponds to will be terminated, even if the bin has not met the minimum number of entries or minimum size.
+ Note that if the FlowFile that triggers the termination of the bin is itself larger than the Maximum Bin Size, it will be placed into its
+ own bin without triggering the termination of any other bin. When using this property, it is recommended to use Prioritizers in the flow's
+ connections to ensure that the ordering is as desired.
+ """)
+ .required(false)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.BOOLEAN, false))
+ .dependsOn(MERGE_STRATEGY, MERGE_STRATEGY_BIN_PACK)
+ .build();
+
+ public static final PropertyDescriptor FLOWFILE_INSERTION_STRATEGY = new PropertyDescriptor.Builder()
+ .name("FlowFile Insertion Strategy")
+ .description("If a given FlowFile terminates the bin based on the <Bin Termination Check> property, specifies where the FlowFile should be included in the bin.")
+ .required(true)
+ .dependsOn(BIN_TERMINATION_CHECK)
+ .defaultValue(InsertionLocation.LAST_IN_BIN)
+ .allowableValues(InsertionLocation.class)
+ .build();
+
+
private static final List<PropertyDescriptor> PROPERTIES = List.of(
MERGE_STRATEGY,
MERGE_FORMAT,
@@ -405,6 +434,8 @@ public class MergeContent extends BinFiles {
addBinPackingDependency(MAX_ENTRIES),
addBinPackingDependency(MIN_SIZE),
addBinPackingDependency(MAX_SIZE),
+ BIN_TERMINATION_CHECK,
+ FLOWFILE_INSERTION_STRATEGY,
MAX_BIN_AGE,
MAX_BIN_COUNT,
DELIMITER_STRATEGY,
@@ -495,6 +526,13 @@ public class MergeContent extends BinFiles {
binManager.setFileCountAttribute(FRAGMENT_COUNT_ATTRIBUTE);
} else {
binManager.setFileCountAttribute(null);
+
+ final PropertyValue terminationCheck = context.getProperty(BIN_TERMINATION_CHECK);
+ if (terminationCheck.isSet()) {
+ final InsertionLocation insertionLocation = context.getProperty(FLOWFILE_INSERTION_STRATEGY).asAllowableValue(InsertionLocation.class);
+ final Predicate<FlowFile> predicate = flowFile -> terminationCheck.evaluateAttributeExpressions(flowFile).asBoolean();
+ binManager.setBinTermination(predicate, insertionLocation);
+ }
}
}
diff --git a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.MergeContent/additionalDetails.md b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.MergeContent/additionalDetails.md
index 2eb50f08ab..dc6a82cf8e 100644
--- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.MergeContent/additionalDetails.md
+++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.MergeContent/additionalDetails.md
@@ -83,7 +83,7 @@ example, the user can specify the minimum number of FlowFiles that must be packa
performed. The minimum number of bytes can also be configured. Additionally, a maximum number of FlowFiles and bytes may
be specified.
-There are also two other conditions that will result in the contents of a Bin being merged together. The Max Bin Age
+There are two other conditions that will result in the contents of a Bin being merged together. The Max Bin Age
property specifies the maximum amount of time that FlowFiles can be binned together before the bin is merged. This
property should almost always be set, as it provides a means to set a timeout on a bin, so that even if data stops
flowing to the Processor for a while (due to a problem with an upstream system, a source processor being stopped, etc.)
@@ -102,6 +102,25 @@ this value or the bin will never be complete. If all the necessary FlowFiles are
which the bin times amount (as specified by the <Max Bin Age> property), then the FlowFiles will all be routed to the '
failure' relationship instead of being merged together.
+Finally, a bin can be merged if the <Bin Termination Check> property is configured and a FlowFile is received that
+satisfies the specified condition. The condition is specified as an Expression Language expression. If any FlowFile
+result in the expression returning a value of true, then the bin will be merged, regardless of how much data is in
+the bin or how old the bin is. This incoming FlowFile that triggers the bin to be merged can either be added as the
+last entry in the bin, as the first entry in a new bin, or output as its own bin, depending on the value of the
+<FlowFile Insertion Strategy> property.
+
+A bin of FlowFiles, then, is merged when any one of the following conditions is met:
+- The bin has reached the maximum number of bytes, as configured by the <Max Group Size> property.
+- The bin has reached the maximum number of FlowFiles, as configured by the <Maximum Number of Entries> property.
+- The bin has reached both the minimum number of bytes, as configured by the <Min Group Size> property,
+ AND the minimum number of FlowFiles, as configured by the <Minimum Number of Entries> property.
+- The bin has reached the maximum age, as configured by the <Max Bin Age> property.
+- The maximum number of bins has been reached, as configured by the <Maximum number of Bins> property, and a new bin must be created.
+- The <Bin Termination Check> property is configured and a FlowFile is received that satisfies the specified condition.
+
+
+### Reason for Merge
+
Whenever the contents of a Bin are merged, an attribute with the name "merge.reason" will be added to the merged
FlowFile. The below table provides a listing of all possible values for this attribute with an explanation of each.
@@ -112,6 +131,7 @@ FlowFile. The below table provides a listing of all possible values for this att
| MIN\_THRESHOLDS\_REACHED | The bin has reached both the minimum number of bytes, as configured by the <Min Group Size> property, AND the minimum number of FlowFiles, as configured by the <Minimum Number of Entries> property. The bin has not reached the maximum number of bytes (Max Group Size) OR the maximum number of FlowFiles (Maximum Number of Entries). [...]
| TIMEOUT | The Bin has reached the maximum age, as configured by the <Max Bin Age> property. If this threshold is reached, the contents of the Bin will be merged together, even if the Bin has not yet reached either of the minimum thresholds. Note that the age here is determined by when the Bin was created, NOT the age of the FlowFiles that reside within those Bins. As a result, if the Processor is stopped until it has 1 million FlowFiles queued, each one being 1 [...]
| BIN\_MANAGER\_FULL | If an incoming FlowFile does not fit into any of the existing Bins (either due to the Maximum thresholds set, or due to the Correlation Attribute being used, etc.), then a new Bin must be created for the incoming FlowFiles. If the number of active Bins is already equal to the <Maximum number of Bins> property, the oldest Bin will be merged in order to make room for the new Bin. In that case, the Bin Manager is said to be full, and this value will be u [...]
+| BIN\_TERMINATION\_SIGNAL | A FlowFile signaled that the Bin should be terminated by satisfying the configured <Bin Termination Check> property. [...]
Note that the attribute value is minimally named, while the textual description is far more verbose. This is done for a
few reasons. Firstly, storing a large value for the attribute can be more costly, utilizing more heap space and
diff --git a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestMergeContent.java b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestMergeContent.java
index 079edc8aa4..e89c7cd95f 100644
--- a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestMergeContent.java
+++ b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestMergeContent.java
@@ -33,6 +33,7 @@ import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.flowfile.attributes.StandardFlowFileMediaType;
import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.bin.InsertionLocation;
import org.apache.nifi.processors.standard.merge.AttributeStrategyUtil;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.MockProcessContext;
@@ -48,6 +49,7 @@ import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -1338,6 +1340,147 @@ public class TestMergeContent {
assertEquals(2, runner.getQueueSize().getObjectCount());
}
+ @Test
+ public void testBinTerminationTriggerStartOfBin() {
+ final TestRunner runner = TestRunners.newTestRunner(new MergeContent());
+ runner.setProperty(MergeContent.MIN_ENTRIES, "10");
+ runner.setProperty(MergeContent.MAX_ENTRIES, "25");
+ runner.setProperty(MergeContent.MAX_BIN_COUNT, "3");
+ runner.setProperty(MergeContent.BIN_TERMINATION_CHECK, "${termination:equals('true')}");
+ runner.setProperty(MergeContent.FLOWFILE_INSERTION_STRATEGY, InsertionLocation.FIRST_IN_NEW_BIN);
+
+ // Enqueue 5 FlowFiles, followed by a FlowFile with the 'termination' attribute set to 'true'.
+ // Do this 3 times.
+ int flowFileIndex = 0;
+ for (int j = 0; j < 3; j++) {
+ for (int i = 0; i < 6; i++) {
+ final Map<String, String> attributes = (i == 5) ? Map.of("termination", "true") : Collections.emptyMap();
+ runner.enqueue((flowFileIndex++) + "\n", attributes);
+ }
+ }
+
+ runner.run(2);
+
+ runner.assertTransferCount(MergeContent.REL_MERGED, 3);
+
+ // We should get out 17 because the last FlowFile has not yet been transferred out, since it is the start of a new bin.
+ runner.assertTransferCount(MergeContent.REL_ORIGINAL, 17);
+
+ final List<MockFlowFile> merged = runner.getFlowFilesForRelationship(MergeContent.REL_MERGED);
+ merged.getFirst().assertContentEquals("0\n1\n2\n3\n4\n");
+ merged.get(1).assertContentEquals("5\n6\n7\n8\n9\n10\n");
+ merged.get(2).assertContentEquals("11\n12\n13\n14\n15\n16\n");
+ }
+
+ @Test
+ public void testBinTerminationTriggerEndOfBin() {
+ final TestRunner runner = TestRunners.newTestRunner(new MergeContent());
+ runner.setProperty(MergeContent.MIN_ENTRIES, "10");
+ runner.setProperty(MergeContent.MAX_ENTRIES, "25");
+ runner.setProperty(MergeContent.MAX_BIN_COUNT, "3");
+ runner.setProperty(MergeContent.BIN_TERMINATION_CHECK, "${termination:equals('true')}");
+ runner.setProperty(MergeContent.FLOWFILE_INSERTION_STRATEGY, InsertionLocation.LAST_IN_BIN);
+
+ // Enqueue 5 FlowFiles, followed by a FlowFile with the 'termination' attribute set to 'true'.
+ // Do this 3 times.
+ int flowFileIndex = 0;
+ for (int j = 0; j < 3; j++) {
+ for (int i = 0; i < 6; i++) {
+ final Map<String, String> attributes = (i == 5) ? Map.of("termination", "true") : Collections.emptyMap();
+ runner.enqueue((flowFileIndex++) + "\n", attributes);
+ }
+ }
+
+ runner.run(2);
+
+ runner.assertTransferCount(MergeContent.REL_MERGED, 3);
+
+ // We should get out 18 because the last FlowFile ended the bin and was transferred out.
+ runner.assertTransferCount(MergeContent.REL_ORIGINAL, 18);
+
+ final List<MockFlowFile> merged = runner.getFlowFilesForRelationship(MergeContent.REL_MERGED);
+ merged.getFirst().assertContentEquals("0\n1\n2\n3\n4\n5\n");
+ merged.get(1).assertContentEquals("6\n7\n8\n9\n10\n11\n");
+ merged.get(2).assertContentEquals("12\n13\n14\n15\n16\n17\n");
+ }
+
+ @Test
+ public void testBinTerminationTriggerIsolated() {
+ final TestRunner runner = TestRunners.newTestRunner(new MergeContent());
+ runner.setProperty(MergeContent.MIN_ENTRIES, "10");
+ runner.setProperty(MergeContent.MAX_ENTRIES, "25");
+ runner.setProperty(MergeContent.MAX_BIN_COUNT, "3");
+ runner.setProperty(MergeContent.BIN_TERMINATION_CHECK, "${termination:equals('true')}");
+ runner.setProperty(MergeContent.FLOWFILE_INSERTION_STRATEGY, InsertionLocation.ISOLATED);
+
+ // Enqueue 5 FlowFiles, followed by a FlowFile with the 'termination' attribute set to 'true'.
+ // Do this 3 times.
+ int flowFileIndex = 0;
+ for (int j = 0; j < 3; j++) {
+ for (int i = 0; i < 6; i++) {
+ final Map<String, String> attributes = (i == 5) ? Map.of("termination", "true") : Collections.emptyMap();
+ runner.enqueue((flowFileIndex++) + "\n", attributes);
+ }
+ }
+
+ runner.run(2);
+
+ runner.assertTransferCount(MergeContent.REL_MERGED, 6);
+
+ // We should get out 18 because the last FlowFile ended the bin and was transferred out.
+ runner.assertTransferCount(MergeContent.REL_ORIGINAL, 18);
+
+ final List<MockFlowFile> merged = runner.getFlowFilesForRelationship(MergeContent.REL_MERGED);
+ merged.getFirst().assertContentEquals("0\n1\n2\n3\n4\n");
+ merged.get(1).assertContentEquals("5\n");
+ merged.get(2).assertContentEquals("6\n7\n8\n9\n10\n");
+ merged.get(3).assertContentEquals("11\n");
+ merged.get(4).assertContentEquals("12\n13\n14\n15\n16\n");
+ merged.get(5).assertContentEquals("17\n");
+ }
+
+ @Test
+ public void testTerminationTriggerWithCorrelationAttribute() {
+ final TestRunner runner = TestRunners.newTestRunner(new MergeContent());
+ runner.setProperty(MergeContent.MIN_ENTRIES, "10");
+ runner.setProperty(MergeContent.MAX_ENTRIES, "25");
+ runner.setProperty(MergeContent.MAX_BIN_COUNT, "3");
+ runner.setProperty(MergeContent.BIN_TERMINATION_CHECK, "${termination:equals('true')}");
+ runner.setProperty(MergeContent.FLOWFILE_INSERTION_STRATEGY, InsertionLocation.FIRST_IN_NEW_BIN);
+ runner.setProperty(MergeContent.CORRELATION_ATTRIBUTE_NAME, "correlation");
+
+ // Enqueue FlowFiles. This should create 2 bins:
+ // '1' should get values 0, 1
+ // Then, '2' should get value 2
+ // Then, FlowFile with content '3' should be the start of a new bin
+ // Then, FlowFile with content '4' should be teh start of another new bin
+ // Then, we add 10 additional FlowFiles to bin 2 in order to fill it without a termination signal
+ // This should leave the FlowFile with content '4' in the bin, but not transferred out.
+ final Map<String, String> terminationAttributes = Map.of("termination", "true", "correlation", "1");
+ final Map<String, String> correlation1 = Map.of("correlation", "1");
+ final Map<String, String> correlation2 = Map.of("correlation", "2");
+
+ runner.enqueue("0\n", correlation1);
+ runner.enqueue("1\n", correlation1);
+ runner.enqueue("2\n", correlation2);
+ runner.enqueue("3\n", terminationAttributes);
+ runner.enqueue("4\n", terminationAttributes);
+
+ for (int i = 0; i < 10; i++) {
+ runner.enqueue(i + "\n", correlation2);
+ }
+
+ runner.run(2);
+
+ runner.assertTransferCount(MergeContent.REL_MERGED, 3);
+ runner.assertTransferCount(MergeContent.REL_ORIGINAL, 14);
+
+ final List<MockFlowFile> merged = runner.getFlowFilesForRelationship(MergeContent.REL_MERGED);
+ merged.getFirst().assertContentEquals("0\n1\n");
+ merged.get(1).assertContentEquals("3\n");
+ merged.get(2).assertContentEquals("2\n0\n1\n2\n3\n4\n5\n6\n7\n8\n9\n");
+ }
+
private void createFlowFiles(final TestRunner testRunner) {
final Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.MIME_TYPE.key(), "application/plain-text");