From 716fd508f4f663d953f48585f2fbcbfbd5aae21e Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Fri, 2 Oct 2026 10:44:25 -0400 Subject: [PATCH] NIFI-16271 Fix Stateless in-memory flow updates Co-authored-by: Cursor --- .../nifi/web/api/dto/ProcessGroupDTO.java | 3 +- .../nifi/groups/StandardProcessGroup.java | 4 +- .../nifi/groups/StandardProcessGroupTest.java | 7 ++- .../org/apache/nifi/groups/ProcessGroup.java | 6 +- .../nifi/web/StandardNiFiServiceFacade.java | 4 ++ .../nifi/web/api/ProcessGroupResource.java | 4 +- .../web/StandardNiFiServiceFacadeTest.java | 55 +++++++++++++++---- .../edit-process-group.component.html | 4 +- .../edit-process-group.component.spec.ts | 2 +- .../edit-process-group.component.ts | 2 +- .../nifi/tests/system/NiFiClientUtil.java | 1 + .../system/stateless/StatelessBasicsIT.java | 19 ++++++- 12 files changed, 84 insertions(+), 27 deletions(-) diff --git a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessGroupDTO.java b/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessGroupDTO.java index 741535b32dff..167759c70015 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/ProcessGroupDTO.java +++ b/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; diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java index 90c0af28b4da..4f8970be64cd 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java +++ b/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); } } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java index 35829aba44f1..dc809dc4eccc 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java +++ b/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()); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java index 145fa71a2b89..d271e773a31b 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/groups/ProcessGroup.java +++ b/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 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 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 */ diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java index 8d8c9766afec..e398e3093d03 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java +++ b/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 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; diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java index 25c3eac431d9..1a90cbded39c 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java +++ b/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); diff --git a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java index 5d7d0f0e5310..d97d59f597a3 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/StandardNiFiServiceFacadeTest.java +++ b/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 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 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 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) { diff --git a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.html b/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.html index b40684426078..e3f0c17df5e4 100644 --- a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.html +++ b/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 @@

{{ 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."> + 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."> diff --git a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.spec.ts b/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.spec.ts index c6365cc30721..27dd05bec27b 100644 --- a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.spec.ts +++ b/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(); diff --git a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.ts b/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.ts index 0b6da9822bf3..353f669310d4 100644 --- a/nifi-frontend/src/main/frontend/apps/nifi/src/app/ui/common/component-dialogs/edit-process-group/edit-process-group.component.ts +++ b/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 }; } diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java index b9324346c8f4..f63d16832acf 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/NiFiClientUtil.java +++ b/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); } diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java index a28361134704..c54e1ae247f4 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/stateless/StatelessBasicsIT.java +++ b/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,6 +822,7 @@ 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)); @@ -822,8 +830,17 @@ public void testChangeFlowVersion() throws NiFiClientException, IOException, Int // 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)); } }