This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 6edc00af55e NIFI-16271 Fix Stateless in-memory flow updates (#11751)
6edc00af55e is described below
commit 6edc00af55ef5835c097519f606c8da343c8f2a4
Author: Mark Payne <[email protected]>
AuthorDate: Fri Oct 2 12:16:08 2026 -0400
NIFI-16271 Fix Stateless in-memory flow updates (#11751)
---
.../apache/nifi/web/api/dto/ProcessGroupDTO.java | 3 +-
.../apache/nifi/groups/StandardProcessGroup.java | 4 +-
.../nifi/groups/StandardProcessGroupTest.java | 7 ++-
.../java/org/apache/nifi/groups/ProcessGroup.java | 6 +--
.../apache/nifi/web/StandardNiFiServiceFacade.java | 4 ++
.../apache/nifi/web/api/ProcessGroupResource.java | 4 +-
.../nifi/web/StandardNiFiServiceFacadeTest.java | 55 +++++++++++++++++-----
.../edit-process-group.component.html | 4 +-
.../edit-process-group.component.spec.ts | 2 +-
.../edit-process-group.component.ts | 2 +-
.../apache/nifi/tests/system/NiFiClientUtil.java | 1 +
.../tests/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 741535b32df..167759c7001 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 class ProcessGroupDTO extends ComponentDTO {
}
@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 90c0af28b4d..4f8970be64c 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 @@ public final class StandardProcessGroup implements
ProcessGroup {
}
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 35829aba44f..dc809dc4ecc 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 @@ class StandardProcessGroupTest {
@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 @@ class StandardProcessGroupTest {
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 145fa71a2b8..d271e773a31 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 @@ public interface ProcessGroup extends
ComponentAuthorizable, Positionable, Versi
/**
* @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 @@ public interface ProcessGroup extends
ComponentAuthorizable, Positionable, Versi
/**
* 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 8d8c9766afe..e398e3093d0 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 class StandardNiFiServiceFacade implements
NiFiServiceFacade {
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 25c3eac431d..1a90cbded39 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 class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
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 5d7d0f0e531..d97d59f597a 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.flowanalysis.EnforcementPolicy;
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.MockBulletinRepository;
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 class StandardNiFiServiceFacadeTest {
}
@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 class StandardNiFiServiceFacadeTest {
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 class StandardNiFiServiceFacadeTest {
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) {
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 b4068442607..e3f0c17df5e 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 @@
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>
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 c6365cc3072..27dd05bec27 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 0b6da9822bf..353f669310d 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 b9324346c8f..f63d16832ac 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 class NiFiClientUtil {
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 a2836113470..c54e1ae247f 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 @@ public class StatelessBasicsIT extends NiFiSystemIT {
@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 class StatelessBasicsIT extends NiFiSystemIT {
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 class StatelessBasicsIT extends NiFiSystemIT {
// 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 class StatelessBasicsIT extends NiFiSystemIT {
// 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));
}
}
