NIFI-16271 Allow Stateless Process Groups to buffer FlowFile content in memory

nifi-docs/src/main/asciidoc/user-guide.adoc

@@ -1781,6 +1781,13 @@ Stateless Engine is not a safe choice if data must be persisted by NiFi. However
 it can be a great choice. Additionally, for protocols such as HTTP, NiFi offers processors that are capable of receiving data, performing some processing, and then sending an acknowledgment.
 As such, it is safe even when NiFi is responsible for accepting incoming connections. The key here is the application-level acknowledgment message that is sent from NiFi.
 
+A Process Group using the Stateless Engine can also be configured with a maximum amount of FlowFile content to keep in memory. This is controlled by two settings: a data size such as `4 GB`,
+and a heap percentage from `0` to `90`. A value of `0 B` or `0` percent means zero -- it does not mean unlimited. If only one setting is configured, that setting is used. If both are
+configured, the smaller of the two limits is used, so `80` percent with a `4 GB` maximum means 80% of the heap, up to 4 GB. If neither is configured, or if the effective limit is zero,
+all content is written to the Content Repository. The default is `0` percent and no data size, which uses no in-memory content. The configured maximum applies to the Stateless Process Group
+as a whole, including all Concurrent Tasks. In-memory content is dropped on restart, just like other data inside a Stateless group. Content that has already spilled to the Content Repository,
+or that has left the group through an Output Port, remains on disk.
+
 
 ==== Data Ordering
 
@@ -1802,7 +1809,8 @@ the data is would still be queued up outside of the Stateless Engine, and this w
 As such, the Provenance Events are not stored into the Provenance Repository until the transaction completes for a Stateless flow. If a FlowFile is routed to a Failure Port,
 or if the invocation times out, the Provenance Events are discard. There are, however, two exceptions to this rule: `SEND` and `REMOTE_INVOCATION` events. Even if the transaction is
 rolled back, the fact that data was sent, or that some remote invocation occurred cannot be rolled back. Therefore, the Provenance Repository is still updated to note the fact that
-these events occurred.
+these events occurred. When FlowFile content is buffered in memory, those Provenance events still refer to that in-memory content, so the content cannot be downloaded or replayed from
+the events after the transaction completes.
 
 
 ==== Site-to-Site

nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessGroupDTO.java

@@ -40,6 +40,8 @@ public class ProcessGroupDTO extends ComponentDTO {
     private String executionEngine;
     private Integer maxConcurrentTasks;
     private String statelessFlowTimeout;
+    private String statelessFlowFileContentInMemoryMax;
+    private String statelessFlowFileContentInMemoryHeapPercentage;
 
     private Integer runningCount;
     private Integer stoppedCount;
@@ -422,4 +424,41 @@ public String getStatelessFlowTimeout() {
     public void setStatelessFlowTimeout(final String timeout) {
         this.statelessFlowTimeout = timeout;
     }
+
+    @Schema(description = "The maximum amount of FlowFile content to buffer in memory when the flow is run using the Stateless Engine, as a data size such as " +
+        "\"4 GB\". A value of \"0 B\" means zero bytes. When this value is not set, only the heap percentage limit is used. When both this value and the heap percentage are set, " +
+        "the smaller of the two limits is used.")
+    public String getStatelessFlowFileContentInMemoryMax() {
+        return statelessFlowFileContentInMemoryMax;
+    }
+
+    public void setStatelessFlowFileContentInMemoryMax(final String statelessFlowFileContentInMemoryMax) {
+        this.statelessFlowFileContentInMemoryMax = statelessFlowFileContentInMemoryMax;
+    }
+
+    @Schema(description = "The maximum percentage of the Java heap to use for buffering FlowFile content when the flow is run using the Stateless Engine, from 0 to 90. " +
+        "A value of 0 means zero percent of the heap. An empty value means this limit is not configured. When both this value and the in-memory content maximum data size are set, " +
+        "the smaller of the two limits is used. The default is 0.")
+    public String getStatelessFlowFileContentInMemoryHeapPercentage() {
+        return statelessFlowFileContentInMemoryHeapPercentage;
+    }
+
+    public void setStatelessFlowFileContentInMemoryHeapPercentage(final String statelessFlowFileContentInMemoryHeapPercentage) {
+        this.statelessFlowFileContentInMemoryHeapPercentage = statelessFlowFileContentInMemoryHeapPercentage;
+    }
+
+    /**
+     * @return the heap percentage as an Integer, or null when the value is blank (not configured)
+     */
+    public Integer toStatelessFlowFileContentInMemoryHeapPercentage() {
+        if (statelessFlowFileContentInMemoryHeapPercentage == null || statelessFlowFileContentInMemoryHeapPercentage.isBlank()) {
+            return null;
+        }
+
+        try {
+            return Integer.valueOf(statelessFlowFileContentInMemoryHeapPercentage.trim());
+        } catch (final NumberFormatException e) {
+            throw new IllegalArgumentException("Illegal value proposed for Max In-Memory Heap Percentage: " + statelessFlowFileContentInMemoryHeapPercentage, e);
+        }
+    }
 }

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java

@@ -501,6 +501,9 @@ private void synchronize(final ProcessGroup group, final VersionedProcessGroup p
         if (statelessTimeout != null) {
             group.setStatelessFlowTimeout(statelessTimeout);
         }
+        final String statelessFlowFileContentInMemoryMax = proposed.getStatelessFlowFileContentInMemoryMax();
+        group.setStatelessContentMaxHeap(statelessFlowFileContentInMemoryMax);
+        group.setStatelessContentMaxHeapPercentage(proposed.getStatelessFlowFileContentInMemoryHeapPercentage());
         if (proposed.getScheduledState() != null && ScheduledState.RUNNING.name().equals(proposed.getScheduledState().name())) {
             context.getComponentScheduler().startStatelessGroup(group);
         }

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java

@@ -128,6 +128,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.net.ConnectException;
 import java.nio.charset.StandardCharsets;
 import java.security.SecureRandom;
@@ -138,6 +140,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -156,6 +159,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -213,6 +217,8 @@ public final class StandardProcessGroup implements ProcessGroup {
     private volatile ExecutionEngine executionEngine = ExecutionEngine.INHERITED;
     private volatile int maxConcurrentTasks = 1;
     private volatile String statelessFlowTimeout = "1 min";
+    private volatile String statelessFlowFileContentInMemoryMax;
+    private volatile Integer statelessFlowFileContentInMemoryHeapPercentage = 0;
     private volatile Authorizable explicitParentAuthorizable;
     private final FlowFileActivity flowFileActivity = new ProcessGroupFlowFileActivity(this);
 
@@ -233,6 +239,7 @@ public final class StandardProcessGroup implements ProcessGroup {
     private static final String DEFAULT_FLOWFILE_EXPIRATION = "0 sec";
     private static final long DEFAULT_BACKPRESSURE_OBJECT = 10_000L;
     private static final String DEFAULT_BACKPRESSURE_DATA_SIZE = "1 GB";
+    private static final int MAX_HEAP_PERCENTAGE = 90;
     private static final Pattern INVALID_DIRECTORY_NAME_CHARACTERS = Pattern.compile("[\\s\\<\\>:\\'\\\"\\/\\\\\\|\\?\\*]");
     private static final String PATH_SEPARATOR = "/";
     private static final String VERSION_SEPARATOR = ":";
@@ -3761,6 +3768,8 @@ private VersionedProcessGroup stripContentsFromRemoteDescendantGroups(final Vers
         copy.setExecutionEngine(processGroup.getExecutionEngine());
         copy.setMaxConcurrentTasks(processGroup.getMaxConcurrentTasks());
         copy.setStatelessFlowTimeout(processGroup.getStatelessFlowTimeout());
+        copy.setStatelessFlowFileContentInMemoryMax(processGroup.getStatelessFlowFileContentInMemoryMax());
+        copy.setStatelessFlowFileContentInMemoryHeapPercentage(processGroup.getStatelessFlowFileContentInMemoryHeapPercentage());
 
         final Set<VersionedProcessGroup> copyChildren = new HashSet<>();
 
@@ -3785,6 +3794,8 @@ private VersionedProcessGroup stripContentsFromRemoteDescendantGroups(final Vers
                 childCopy.setExecutionEngine(childGroup.getExecutionEngine());
                 childCopy.setMaxConcurrentTasks(childGroup.getMaxConcurrentTasks());
                 childCopy.setStatelessFlowTimeout(childGroup.getStatelessFlowTimeout());
+                childCopy.setStatelessFlowFileContentInMemoryMax(childGroup.getStatelessFlowFileContentInMemoryMax());
+                childCopy.setStatelessFlowFileContentInMemoryHeapPercentage(childGroup.getStatelessFlowFileContentInMemoryHeapPercentage());
 
                 copyChildren.add(childCopy);
             }
@@ -4778,6 +4789,128 @@ public void setStatelessFlowTimeout(final String statelessFlowTimeout) {
         }
     }
 
+    @Override
+    public String getStatelessContentMaxHeap() {
+        return statelessFlowFileContentInMemoryMax;
+    }
+
+    @Override
+    public void setStatelessContentMaxHeap(final String maxSize) {
+        writeLock.lock();
+        try {
+            verifyCanSetStatelessContentMaxHeap(maxSize);
+            this.statelessFlowFileContentInMemoryMax = normalizeStatelessContentMaxHeap(maxSize);
+        } finally {
+            writeLock.unlock();
+        }
+    }
+
+    @Override
+    public Integer getStatelessContentMaxHeapPercentage() {
+        return statelessFlowFileContentInMemoryHeapPercentage;
+    }
+
+    @Override
+    public void setStatelessContentMaxHeapPercentage(final Integer heapPercentage) {
+        writeLock.lock();
+        try {
+            verifyCanSetStatelessContentMaxHeapPercentage(heapPercentage);
+            this.statelessFlowFileContentInMemoryHeapPercentage = heapPercentage;
+        } finally {
+            writeLock.unlock();
+        }
+    }
+
+    @Override
+    public long resolveStatelessContentMaxHeap() {
+        return resolveStatelessContentMaxHeap(getStatelessContentMaxHeap(), getStatelessContentMaxHeapPercentage());
+    }
+
+    @Override
+    public void verifyCanSetStatelessContentMaxHeap(final String maxSize) {
+        final long proposedMaxSizeBytes = resolveStatelessContentMaxHeap(maxSize, getStatelessContentMaxHeapPercentage());
+        verifyCanSetStatelessContentMaxHeap(proposedMaxSizeBytes);
+    }
+
+    @Override
+    public void verifyCanSetStatelessContentMaxHeapPercentage(final Integer heapPercentage) {
+        final long proposedMaxSizeBytes = resolveStatelessContentMaxHeap(getStatelessContentMaxHeap(), heapPercentage);
+        verifyCanSetStatelessContentMaxHeap(proposedMaxSizeBytes);
+    }
+
+    private void verifyCanSetStatelessContentMaxHeap(final long proposedMaxSizeBytes) {
+        // The Content Repository is selected when the Stateless flow starts, so the setting cannot change while the flow is running.
+        final ProcessGroup statelessGroup = getStatelessGroup(this);
+        if (statelessGroup != null && statelessGroup.getStatelessScheduledState() != StatelessGroupScheduledState.STOPPED
+            && proposedMaxSizeBytes != resolveStatelessContentMaxHeap()) {
+            throw new IllegalStateException("Cannot change the maximum in-memory FlowFile content for " + this
+                + " while the Stateless flow is running. Stop the Process Group before changing this setting.");
+        }
+    }
+
+    private static String normalizeStatelessContentMaxHeap(final String maxSize) {
+        if (maxSize == null || maxSize.isBlank()) {
+            return null;
+        }
+
+        return maxSize.trim();
+    }
+
+    private static long resolveStatelessContentMaxHeap(final String maxSize, final Integer heapPercentage) {
+        validateHeapPercentage(heapPercentage);
+
+        final boolean sizeConfigured = maxSize != null && !maxSize.isBlank();
+        final boolean heapPercentageConfigured = heapPercentage != null;
+        if (!sizeConfigured && !heapPercentageConfigured) {
+            return 0L;
+        }
+
+        if (sizeConfigured && !heapPercentageConfigured) {
+            return parseStatelessContentMaxHeap(maxSize);
+        }
+        if (!sizeConfigured) {
+            return toHeapPercentageBytes(heapPercentage);
+        }
+
+        return Math.min(parseStatelessContentMaxHeap(maxSize), toHeapPercentageBytes(heapPercentage));
+    }
+
+    private static void validateHeapPercentage(final Integer heapPercentage) {
+        if (heapPercentage != null && (heapPercentage < 0 || heapPercentage > MAX_HEAP_PERCENTAGE)) {
+            throw new IllegalArgumentException("Heap percentage must be between 0 and " + MAX_HEAP_PERCENTAGE + ": " + heapPercentage);
+        }
+    }
+
+    private static long toHeapPercentageBytes(final Integer heapPercentage) {
+        return BigDecimal.valueOf(heapPercentage)
+            .multiply(BigDecimal.valueOf(Runtime.getRuntime().maxMemory()))
+            .divide(BigDecimal.valueOf(100), 0, RoundingMode.DOWN)
+            .longValue();
+    }
+
+    private static long parseStatelessContentMaxHeap(final String maxSize) {
+        final String normalizedMaxSize = maxSize.trim().toUpperCase(Locale.ROOT);
+        final Matcher matcher = DataUnit.DATA_SIZE_PATTERN.matcher(normalizedMaxSize);
+        if (!matcher.matches()) {
+            throw new IllegalArgumentException("Invalid data size: " + maxSize);
+        }
+
+        final long multiplier = switch (matcher.group(2)) {
+            case "B" -> 1L;
+            case "KB" -> 1L << 10;
+            case "MB" -> 1L << 20;
+            case "GB" -> 1L << 30;
+            case "TB" -> 1L << 40;
+            default -> throw new IllegalArgumentException("Invalid data size: " + maxSize);
+        };
+
+        try {
+            return new BigDecimal(matcher.group(1)).multiply(BigDecimal.valueOf(multiplier)).longValueExact();
+        } catch (final ArithmeticException e) {
+            throw new IllegalArgumentException("Data size must represent a whole number of bytes within the supported range: " + maxSize, e);
+        }
+    }
+
     private void setLoggingAttributes(final boolean recursive) {
         final Map<String, String> attributes = new HashMap<>();
 

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/registry/flow/mapping/VersionedComponentFlowMapper.java

@@ -274,6 +274,8 @@ private InstantiatedVersionedProcessGroup mapGroup(final ProcessGroup group, fin
         versionedGroup.setScheduledState(flowMappingOptions.getStateLookup().getState(group));
         versionedGroup.setMaxConcurrentTasks(group.getMaxConcurrentTasks());
         versionedGroup.setStatelessFlowTimeout(group.getStatelessFlowTimeout());
+        versionedGroup.setStatelessFlowFileContentInMemoryMax(group.getStatelessContentMaxHeap());
+        versionedGroup.setStatelessFlowFileContentInMemoryHeapPercentage(group.getStatelessContentMaxHeapPercentage());
 
         final ParameterContext parameterContext = group.getParameterContext();
         versionedGroup.setParameterContextName(parameterContext == null ? null : parameterContext.getName());