NIFI-16271 Fix Stateless in-memory flow updates

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

@@ -449,7 +449,8 @@ public void setStatelessFlowFileContentInMemoryMax(final String statelessFlowFil
     }
 
     @Schema(description = "The maximum percentage of the Java heap to use for buffering FlowFile content when the flow is run using the Stateless Engine. " +
-        "The value must be greater than 0 and no more than 90. An empty value means this limit is not configured. When both this value and the in-memory content maximum data size " +
+        "The value must be at least 0 and no more than 90. A value of 0 means zero percent. 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.")
     public String getStatelessFlowFileContentInMemoryHeapPercentage() {
         return statelessFlowFileContentInMemoryHeapPercentage;

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

@@ -4881,8 +4881,8 @@ private static long resolveStatelessContentMaxHeap(final String maxSize, final I
     }
 
     private static void validateHeapPercentage(final Integer heapPercentage) {
-        if (heapPercentage != null && (heapPercentage <= 0 || heapPercentage > MAX_HEAP_PERCENTAGE)) {
-            throw new IllegalArgumentException("Heap percentage must be greater than 0 and no more than " + MAX_HEAP_PERCENTAGE + ": " + heapPercentage);
+        if (heapPercentage != null && (heapPercentage < 0 || heapPercentage > MAX_HEAP_PERCENTAGE)) {
+            throw new IllegalArgumentException("Heap percentage must be at least 0 and no more than " + MAX_HEAP_PERCENTAGE + ": " + heapPercentage);
         }
     }
 

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java

@@ -511,6 +511,10 @@ void testSetStatelessContentMaxHeapParsesDataSize() {
 
     @Test
     void testSetStatelessContentMaxHeapPercentage() {
+        processGroup.setStatelessContentMaxHeapPercentage(0);
+        assertEquals(0, processGroup.getStatelessContentMaxHeapPercentage());
+        assertEquals(0L, processGroup.resolveStatelessContentMaxHeap());
+
         processGroup.setStatelessContentMaxHeapPercentage(50);
         assertEquals(50, processGroup.getStatelessContentMaxHeapPercentage());
         assertEquals(Runtime.getRuntime().maxMemory() / 2, processGroup.resolveStatelessContentMaxHeap());
@@ -544,11 +548,10 @@ void testSetStatelessContentMaxHeapRejectsInvalidLimits() {
         assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeap("0.9 B"));
         assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeap("999999999999999999999999999999999999999999999 TB"));
         assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeap("50%"));
-        assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeapPercentage(0));
         assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeapPercentage(91));
         assertThrows(IllegalArgumentException.class, () -> processGroup.setStatelessContentMaxHeapPercentage(-1));
         assertThrows(IllegalArgumentException.class, () -> processGroup.verifyCanSetStatelessContentMaxHeap("not a size"));
-        assertThrows(IllegalArgumentException.class, () -> processGroup.verifyCanSetStatelessContentMaxHeapPercentage(0));
+        processGroup.verifyCanSetStatelessContentMaxHeapPercentage(0);
         assertEquals("1 MB", processGroup.getStatelessContentMaxHeap());
         assertNull(processGroup.getStatelessContentMaxHeapPercentage());
         assertEquals(1024L * 1024L, processGroup.resolveStatelessContentMaxHeap());

nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java

@@ -1340,13 +1340,13 @@ default void setConnectorLoggingAttributes(final Map<String, String> attributes)
 
     /**
      * @return the configured maximum percentage of the Java heap to use for buffering FlowFile content when this Process Group is run using the Stateless
-     * Execution Engine. The value is greater than 0 and no more than 90. A null value means this limit is not configured.
+     * Execution Engine. The value is at least 0 and no more than 90. A value of 0 means zero percent. A null value means this limit is not configured.
      */
     Integer getStatelessContentMaxHeapPercentage();
 
     /**
      * Sets the maximum percentage of the Java heap to use for buffering FlowFile content when this Process Group is run using the Stateless Execution Engine
-     * @param heapPercentage the maximum heap percentage, greater than 0 and no more than 90. A null value means this limit is not configured.
+     * @param heapPercentage the maximum heap percentage, at least 0 and no more than 90. A null value means this limit is not configured.
      */
     void setStatelessContentMaxHeapPercentage(Integer heapPercentage);
 
@@ -1367,7 +1367,7 @@ default void setConnectorLoggingAttributes(final Map<String, String> attributes)
 
     /**
      * Verifies that the maximum in-memory FlowFile content heap percentage can be set to the given value.
-     * @param heapPercentage the maximum heap percentage, greater than 0 and no more than 90. A null value means this limit is not configured.
+     * @param heapPercentage the maximum heap percentage, at least 0 and no more than 90. A null value means this limit is not configured.
      * @throws IllegalArgumentException if the value is outside the allowed range
      * @throws IllegalStateException if the value cannot be set because the Stateless flow is running
      */

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java

@@ -7311,6 +7311,10 @@ public Set<AffectedComponentEntity> getComponentsAffectedByFlowUpdate(final Stri
                                 state = outputPort.getScheduledState().name();
                             }
                             break;
+                        case PROCESS_GROUP:
+                            localGroup = processGroupDAO.getProcessGroup(localComponent.getInstanceIdentifier());
+                            state = localGroup.getStatelessScheduledState().name();
+                            break;
                         default:
                             state = null;
                             localGroup = null;

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java

@@ -594,8 +594,8 @@ public Response updateProcessGroup(
         if (heapPercentage != null && !heapPercentage.isBlank()) {
             try {
                 final int parsedHeapPercentage = Integer.parseInt(heapPercentage.trim());
-                if (parsedHeapPercentage <= 0 || parsedHeapPercentage > 90) {
-                    throw new IllegalArgumentException("Max In-Memory Heap Percentage must be greater than 0 and no more than 90: " + heapPercentage);
+                if (parsedHeapPercentage < 0 || parsedHeapPercentage > 90) {
+                    throw new IllegalArgumentException("Max In-Memory Heap Percentage must be at least 0 and no more than 90: " + heapPercentage);
                 }
             } catch (final NumberFormatException e) {
                 throw new IllegalArgumentException("Max In-Memory Heap Percentage must be a whole number: " + heapPercentage);

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java

@@ -85,6 +85,7 @@
 import org.apache.nifi.groups.ComponentAdditions;
 import org.apache.nifi.groups.ProcessGroup;
 import org.apache.nifi.groups.RemoteProcessGroup;
+import org.apache.nifi.groups.StatelessGroupScheduledState;
 import org.apache.nifi.groups.VersionedComponentAdditions;
 import org.apache.nifi.history.History;
 import org.apache.nifi.history.HistoryQuery;
@@ -130,6 +131,7 @@
 import org.apache.nifi.util.NiFiProperties;
 import org.apache.nifi.validation.RuleViolation;
 import org.apache.nifi.validation.RuleViolationsManager;
+import org.apache.nifi.web.api.dto.AffectedComponentDTO;
 import org.apache.nifi.web.api.dto.BacklogDTO;
 import org.apache.nifi.web.api.dto.BulletinBoardDTO;
 import org.apache.nifi.web.api.dto.BulletinQueryDTO;
@@ -406,14 +408,16 @@ public Resource getResource() {
     }
 
     @Test
-    public void testGetComponentsAffectedByFlowUpdate_WithNewStatelessProcessGroup_ReproducesNPE() {
+    public void testGetComponentsAffectedByFlowUpdateForStatelessGroups() {
         final String groupId = UUID.randomUUID().toString();
         final ProcessGroup processGroup = mock(ProcessGroup.class);
         when(processGroupDAO.getProcessGroup(groupId)).thenReturn(processGroup);
         final FlowManager flowManager = mock(FlowManager.class);
         final ExtensionManager extensionManager = mock(ExtensionManager.class);
+        final ControllerServiceProvider controllerServiceProvider = mock(ControllerServiceProvider.class);
         when(flowController.getFlowManager()).thenReturn(flowManager);
         when(flowController.getExtensionManager()).thenReturn(extensionManager);
+        when(flowController.getControllerServiceProvider()).thenReturn(controllerServiceProvider);
 
         final StandardNiFiServiceFacade serviceFacadeSpy = spy(serviceFacade);
         final VersionedComponentFlowMapper flowMapper = mock(VersionedComponentFlowMapper.class);
@@ -423,7 +427,6 @@ public void testGetComponentsAffectedByFlowUpdate_WithNewStatelessProcessGroup_R
         when(flowMapper.mapProcessGroup(any(ProcessGroup.class), any(ControllerServiceProvider.class), any(FlowManager.class), eq(true)))
                 .thenReturn(localRoot);
 
-        // Build proposed (updated) flow with a NEW child Process Group configured with Stateless Execution Engine
         final VersionedProcessGroup proposedRoot = new VersionedProcessGroup();
         proposedRoot.setIdentifier("root");
         proposedRoot.setName("root");
@@ -450,16 +453,44 @@ public void testGetComponentsAffectedByFlowUpdate_WithNewStatelessProcessGroup_R
 
         final FlowComparison comparison = flowComparator.compare();
         final boolean hasExecEngineChange = comparison.getDifferences().stream()
-                .anyMatch(d -> d.getDifferenceType() == DifferenceType.EXECUTION_ENGINE_CHANGED
-                        && d.getComponentA() == null
-                        && d.getComponentB() instanceof VersionedProcessGroup
-                        && "child".equals(d.getComponentB().getIdentifier()));
-        assertTrue(hasExecEngineChange, "Expected EXECUTION_ENGINE_CHANGED difference for Stateless child group");
-
-        // Act: Should not throw after fix; no local components are affected by a new Stateless child group
-        final Set<AffectedComponentEntity> affected = serviceFacadeSpy.getComponentsAffectedByFlowUpdate(groupId, updatedSnapshot);
-        assertNotNull(affected);
-        assertTrue(affected.isEmpty(), "No local components should be affected for added Stateless group");
+                .anyMatch(difference -> difference.getDifferenceType() == DifferenceType.EXECUTION_ENGINE_CHANGED
+                        && difference.getComponentA() == null
+                        && difference.getComponentB() instanceof VersionedProcessGroup
+                        && "child".equals(difference.getComponentB().getIdentifier()));
+        assertTrue(hasExecEngineChange);
+
+        final Set<AffectedComponentEntity> affectedByAddedGroup = serviceFacadeSpy.getComponentsAffectedByFlowUpdate(groupId, updatedSnapshot);
+        assertNotNull(affectedByAddedGroup);
+        assertTrue(affectedByAddedGroup.isEmpty());
+
+        final InstantiatedVersionedProcessGroup localStatelessGroup = new InstantiatedVersionedProcessGroup(groupId, "parent");
+        localStatelessGroup.setIdentifier("root");
+        localStatelessGroup.setName("root");
+        localStatelessGroup.setExecutionEngine(ExecutionEngine.STATELESS);
+        localStatelessGroup.setStatelessFlowFileContentInMemoryHeapPercentage(10);
+        when(flowMapper.mapProcessGroup(any(ProcessGroup.class), any(ControllerServiceProvider.class), any(FlowManager.class), eq(true)))
+                .thenReturn(localStatelessGroup);
+
+        final VersionedProcessGroup proposedStatelessGroup = new VersionedProcessGroup();
+        proposedStatelessGroup.setIdentifier("root");
+        proposedStatelessGroup.setName("root");
+        proposedStatelessGroup.setExecutionEngine(ExecutionEngine.STATELESS);
+        proposedStatelessGroup.setStatelessFlowFileContentInMemoryHeapPercentage(20);
+        updatedSnapshot.setFlowContents(proposedStatelessGroup);
+
+        when(processGroup.getIdentifier()).thenReturn(groupId);
+        when(processGroup.getProcessGroupIdentifier()).thenReturn("parent");
+        when(processGroup.resolveExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+        when(processGroup.getStatelessScheduledState()).thenReturn(StatelessGroupScheduledState.RUNNING);
+        serviceFacadeSpy.setRevisionManager(new NaiveRevisionManager());
+        serviceFacadeSpy.setDtoFactory(mock(DtoFactory.class));
+        serviceFacadeSpy.setAuthorizableLookup(mock(AuthorizableLookup.class, Answers.RETURNS_DEEP_STUBS));
+
+        final Set<AffectedComponentEntity> affectedByLimitChange = serviceFacadeSpy.getComponentsAffectedByFlowUpdate(groupId, updatedSnapshot);
+        assertEquals(1, affectedByLimitChange.size());
+        final AffectedComponentEntity affectedGroup = affectedByLimitChange.iterator().next();
+        assertEquals(AffectedComponentDTO.COMPONENT_TYPE_STATELESS_GROUP, affectedGroup.getComponent().getReferenceType());
+        assertEquals(StatelessGroupScheduledState.RUNNING.name(), affectedGroup.getComponent().getState());
     }
 
     private FlowChangeAction getAction(final Integer actionId, final String processorId) {

nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.html

@@ -162,13 +162,13 @@ <h2 mat-dialog-title>{{ readonly ? 'Process Group Details' : 'Edit Process Group
                                             class="fa fa-info-circle"
                                             nifiTooltip
                                             [tooltipComponentType]="TextTip"
-                                            tooltipInputData="A whole number greater than 0 and no more than 90. Leave empty if this limit should not apply. When both this value and the data size are set, the smaller of the two is used. For example, 80% up to 4 GB."></i>
+                                            tooltipInputData="A whole number from 0 through 90. 0 means no FlowFile content is kept in memory. Leave empty if this limit should not apply. When both this value and the data size are set, the smaller of the two is used. For example, 80% up to 4 GB."></i>
                                     </mat-label>
                                     <input
                                         matInput
                                         formControlName="statelessFlowFileContentInMemoryHeapPercentage"
                                         type="number"
-                                        min="1"
+                                        min="0"
                                         max="90"
                                         [readonly]="readonly" />
                                 </mat-form-field>

nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.spec.ts

@@ -188,7 +188,7 @@ describe('EditProcessGroup', () => {
             const control = component.editProcessGroupForm.get('statelessFlowFileContentInMemoryHeapPercentage');
 
             control?.setValue(0);
-            expect(control?.valid).toBeFalsy();
+            expect(control?.valid).toBeTruthy();
 
             control?.setValue(1);
             expect(control?.valid).toBeTruthy();

nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.ts

@@ -333,7 +333,7 @@ export class EditProcessGroup extends TabbedDialog {
         }
 
         const percentage = Number(control.value);
-        if (!Number.isInteger(percentage) || percentage <= 0 || percentage > EditProcessGroup.MAXIMUM_HEAP_PERCENTAGE) {
+        if (!Number.isInteger(percentage) || percentage < 0 || percentage > EditProcessGroup.MAXIMUM_HEAP_PERCENTAGE) {
             return { heapPercentage: true };
         }
 

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java

@@ -2912,6 +2912,7 @@ public ProcessGroupEntity setStatelessFlowFileContentInMemoryMax(final ProcessGr
         final ProcessGroupEntity current = nifiClient.getProcessGroupClient().getProcessGroup(group.getId());
         current.getComponent().setStatelessFlowFileContentInMemoryMax(inMemoryContentMax);
         current.getComponent().setStatelessFlowFileContentInMemoryHeapPercentage("");
+        current.getComponent().setVersionControlInformation(null);
         return nifiClient.getProcessGroupClient().updateProcessGroup(current);
     }
 

nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java

@@ -786,6 +786,7 @@ private ProvenanceSearchValueDTO createSearchValueDto(final String value) {
     @Test
     public void testChangeFlowVersion() throws NiFiClientException, IOException, InterruptedException {
         createFlowShell();
+        statelessGroup = getClientUtil().setStatelessFlowFileContentInMemoryMax(statelessGroup, "1 MB");
         final ConnectionEntity inputToOutput = getClientUtil().createConnection(inputPort, outputPort, statelessGroup.getId());
 
         final FlowRegistryClientEntity registryClient = registerClient();
@@ -800,9 +801,15 @@ public void testChangeFlowVersion() throws NiFiClientException, IOException, Int
         final ProcessorEntity reverseContents = getClientUtil().createProcessor(REVERSE_CONTENTS, statelessGroup.getId());
         getClientUtil().createConnection(inputPort, reverseContents, statelessGroup.getId());
         getClientUtil().createConnection(reverseContents, outputPort, SUCCESS, statelessGroup.getId());
+        statelessGroup = getClientUtil().setStatelessFlowFileContentInMemoryMax(statelessGroup, "2 MB");
 
         // Save v2 of the flow
-        getClientUtil().saveFlowVersion(statelessGroup, registryClient, vci);
+        final VersionControlInformationEntity secondVersionInformation = getClientUtil().saveFlowVersion(statelessGroup, registryClient, vci);
+        waitFor(() -> VersionControlInformationDTO.UP_TO_DATE.equals(getClientUtil().getVersionControlState(statelessGroup.getId())));
+
+        // Save v3 with only the in-memory limit changed
+        statelessGroup = getClientUtil().setStatelessFlowFileContentInMemoryMax(statelessGroup, "3 MB");
+        getClientUtil().saveFlowVersion(statelessGroup, registryClient, secondVersionInformation);
         waitFor(() -> VersionControlInformationDTO.UP_TO_DATE.equals(getClientUtil().getVersionControlState(statelessGroup.getId())));
 
         // Let a FlowFile go through and verify the results
@@ -815,15 +822,25 @@ public void testChangeFlowVersion() throws NiFiClientException, IOException, Int
 
         // Switch back to v1 while flow is running
         getClientUtil().changeFlowVersion(statelessGroup.getId(), "1");
+        assertEquals("1 MB", getNifiClient().getProcessGroupClient().getProcessGroup(statelessGroup.getId()).getComponent().getStatelessFlowFileContentInMemoryMax());
         getClientUtil().startProcessor(generate);
         waitForQueueCount(outputToTerminate, 2);
         assertEquals(HELLO_WORLD, getClientUtil().getFlowFileContentAsUtf8(outputToTerminate.getId(), 1));
         getClientUtil().stopProcessor(generate);
 
         // Switch back to v2 while flow is running
         getClientUtil().changeFlowVersion(statelessGroup.getId(), "2");
+        assertEquals("2 MB", getNifiClient().getProcessGroupClient().getProcessGroup(statelessGroup.getId()).getComponent().getStatelessFlowFileContentInMemoryMax());
         getClientUtil().startProcessor(generate);
         waitForQueueCount(outputToTerminate, 3);
         assertEquals(HELLO_WORLD_REVERSED, getClientUtil().getFlowFileContentAsUtf8(outputToTerminate.getId(), 2));
+        getClientUtil().stopProcessor(generate);
+
+        // Switch to v3 while the flow is running. Only the in-memory limit differs from v2.
+        getClientUtil().changeFlowVersion(statelessGroup.getId(), "3");
+        assertEquals("3 MB", getNifiClient().getProcessGroupClient().getProcessGroup(statelessGroup.getId()).getComponent().getStatelessFlowFileContentInMemoryMax());
+        getClientUtil().startProcessor(generate);
+        waitForQueueCount(outputToTerminate, 4);
+        assertEquals(HELLO_WORLD_REVERSED, getClientUtil().getFlowFileContentAsUtf8(outputToTerminate.getId(), 3));
     }
 }