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 244895a3e30 NIFI-16413 Skip starting processors in stateless groups
(#11746)
244895a3e30 is described below
commit 244895a3e309f611daeec5fdfb308eda1a5059ac
Author: Mark Payne <[email protected]>
AuthorDate: Fri Oct 2 04:15:57 2026 -0400
NIFI-16413 Skip starting processors in stateless groups (#11746)
---
.../StandaloneProcessGroupLifecycle.java | 25 ++++++--
.../TestStandaloneProcessGroupLifecycle.java | 72 +++++++++++++++++++---
.../controller/service/mock/MockProcessGroup.java | 40 +++++++++---
3 files changed, 114 insertions(+), 23 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
index fcc078f3c74..ac09ec572ce 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
@@ -198,10 +198,17 @@ public class StandaloneProcessGroupLifecycle implements
ProcessGroupLifecycle {
@Override
public CompletableFuture<Void> startProcessors(final
ComponentHierarchyScope scope) {
- final boolean recursive = (scope ==
ComponentHierarchyScope.INCLUDE_CHILD_GROUPS);
- final Collection<ProcessorNode> processors = recursive ?
processGroup.findAllProcessors() : processGroup.getProcessors();
+ if (processGroup.resolveExecutionEngine() ==
ExecutionEngine.STATELESS) {
+ return CompletableFuture.completedFuture(null);
+ }
+
final List<CompletableFuture<Void>> startFutures = new ArrayList<>();
- for (final ProcessorNode processor : processors) {
+ startProcessors(processGroup, scope, startFutures);
+ return CompletableFuture.allOf(startFutures.toArray(new
CompletableFuture[0]));
+ }
+
+ private void startProcessors(final ProcessGroup startGroup, final
ComponentHierarchyScope scope, final List<CompletableFuture<Void>>
startFutures) {
+ for (final ProcessorNode processor : startGroup.getProcessors()) {
// If Processor is not valid, perform validation again to ensure
that the status is up to date.
final ValidationStatus validationStatus =
processor.getValidationStatus();
if (validationStatus != ValidationStatus.VALID) {
@@ -213,10 +220,18 @@ public class StandaloneProcessGroupLifecycle implements
ProcessGroupLifecycle {
continue;
}
-
startFutures.add(processor.getProcessGroup().startProcessor(processor, true));
+ startFutures.add(startGroup.startProcessor(processor, true));
}
- return CompletableFuture.allOf(startFutures.toArray(new
CompletableFuture[0]));
+ if (scope == ComponentHierarchyScope.INCLUDE_CHILD_GROUPS) {
+ for (final ProcessGroup childGroup :
startGroup.getProcessGroups()) {
+ if (childGroup.getExecutionEngine() ==
ExecutionEngine.STATELESS) {
+ continue;
+ }
+
+ startProcessors(childGroup,
ComponentHierarchyScope.INCLUDE_CHILD_GROUPS, startFutures);
+ }
+ }
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/facades/standalone/TestStandaloneProcessGroupLifecycle.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/facades/standalone/TestStandaloneProcessGroupLifecycle.java
index 6aa4670aa5d..0f03c83237d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/facades/standalone/TestStandaloneProcessGroupLifecycle.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/facades/standalone/TestStandaloneProcessGroupLifecycle.java
@@ -21,11 +21,16 @@ 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.connector.components.ComponentHierarchyScope;
import org.apache.nifi.components.connector.components.StatelessGroupLifecycle;
+import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ProcessorNode;
+import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.controller.service.mock.MockProcessGroup;
+import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.parameter.ParameterLookup;
import org.junit.jupiter.api.BeforeEach;
@@ -33,6 +38,8 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -46,11 +53,14 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
class TestStandaloneProcessGroupLifecycle {
private static final String SERVICE_ID = "service-1";
@@ -94,7 +104,43 @@ class TestStandaloneProcessGroupLifecycle {
@BeforeEach
void setUp() {
lifecycle = new StandaloneProcessGroupLifecycle(processGroup,
controllerServiceProvider, statelessGroupLifecycle, id -> null);
-
lenient().when(processGroup.getProcessGroups()).thenReturn(Collections.emptySet());
+
when(processGroup.getProcessGroups()).thenReturn(Collections.emptySet());
+ }
+
+ @Test
+ void testStartProcessorsIncludesStandardGroupsOnly() throws Exception {
+ final MockProcessGroup rootGroup = spy(new MockProcessGroup(null,
"root"));
+ final MockProcessGroup standardChildGroup = spy(new
MockProcessGroup(null, "standard"));
+ final MockProcessGroup statelessChildGroup = spy(new
MockProcessGroup(null, "stateless"));
+ final MockProcessGroup inheritedStatelessChildGroup = spy(new
MockProcessGroup(null, "inherited-stateless"));
+ final ProcessorNode rootProcessor =
createStartableProcessor(rootGroup);
+ final ProcessorNode standardChildProcessor =
createStartableProcessor(standardChildGroup);
+ final ProcessorNode statelessChildProcessor =
createStartableProcessor(statelessChildGroup);
+ final ProcessorNode inheritedStatelessChildProcessor =
createStartableProcessor(inheritedStatelessChildGroup);
+
+ rootGroup.setExecutionEngine(ExecutionEngine.STANDARD);
+ standardChildGroup.setExecutionEngine(ExecutionEngine.INHERITED);
+ statelessChildGroup.setExecutionEngine(ExecutionEngine.STATELESS);
+
inheritedStatelessChildGroup.setExecutionEngine(ExecutionEngine.INHERITED);
+
+ rootGroup.addProcessor(rootProcessor);
+ standardChildGroup.addProcessor(standardChildProcessor);
+ statelessChildGroup.addProcessor(statelessChildProcessor);
+
inheritedStatelessChildGroup.addProcessor(inheritedStatelessChildProcessor);
+ statelessChildGroup.addProcessGroup(inheritedStatelessChildGroup);
+ rootGroup.addProcessGroup(standardChildGroup);
+ rootGroup.addProcessGroup(statelessChildGroup);
+
+ final StandaloneProcessGroupLifecycle rootLifecycle = new
StandaloneProcessGroupLifecycle(
+ rootGroup, controllerServiceProvider, statelessGroupLifecycle, id
-> null);
+
+
rootLifecycle.startProcessors(ComponentHierarchyScope.INCLUDE_CHILD_GROUPS).get();
+
+ verify(rootGroup).startProcessor(rootProcessor, true);
+ verify(standardChildGroup).startProcessor(standardChildProcessor,
true);
+ verify(statelessChildGroup,
never()).startProcessor(statelessChildProcessor, true);
+ verify(inheritedStatelessChildGroup,
never()).startProcessor(inheritedStatelessChildProcessor, true);
+ verify(rootGroup, never()).findAllProcessors();
}
@Test
@@ -244,6 +290,16 @@ class TestStandaloneProcessGroupLifecycle {
assertEquals(1, result.size());
}
+ private ProcessorNode createStartableProcessor(final ProcessGroup
owningGroup) {
+ final ProcessorNode processor = mock(ProcessorNode.class);
+ final String processorIdentifier = owningGroup.getIdentifier() +
"-processor";
+ when(processor.getIdentifier()).thenReturn(processorIdentifier);
+
when(processor.getValidationStatus()).thenReturn(ValidationStatus.VALID);
+ when(processor.getScheduledState()).thenReturn(ScheduledState.STOPPED);
+ when(processor.getProcessGroup()).thenReturn(owningGroup);
+ return processor;
+ }
+
/**
* Creates a mock ProcessorNode with the given property descriptors and
effective property values.
* The mock is configured with a ValidationContext whose
isDependencySatisfied calls the real default
@@ -255,8 +311,8 @@ class TestStandaloneProcessGroupLifecycle {
when(processor.getPropertyDescriptors()).thenReturn(descriptors);
for (final PropertyDescriptor descriptor : descriptors) {
-
lenient().when(processor.getPropertyDescriptor(descriptor.getName())).thenReturn(descriptor);
-
lenient().when(processor.getEffectivePropertyValue(descriptor)).thenReturn(effectiveValues.get(descriptor));
+
when(processor.getPropertyDescriptor(descriptor.getName())).thenReturn(descriptor);
+
when(processor.getEffectivePropertyValue(descriptor)).thenReturn(effectiveValues.get(descriptor));
}
// Build the effective property value map for creating the validation
context
@@ -278,13 +334,13 @@ class TestStandaloneProcessGroupLifecycle {
final String value = effectiveValues.get(descriptor);
if (value != null) {
final PropertyValue propertyValue = mock(PropertyValue.class);
- lenient().when(propertyValue.getValue()).thenReturn(value);
-
lenient().when(validationContext.getProperty(descriptor)).thenReturn(propertyValue);
+ when(propertyValue.getValue()).thenReturn(value);
+
when(validationContext.getProperty(descriptor)).thenReturn(propertyValue);
}
}
- lenient().when(processor.getAnnotationData()).thenReturn(null);
-
lenient().when(processor.getParameterLookup()).thenReturn(ParameterLookup.EMPTY);
+ when(processor.getAnnotationData()).thenReturn(null);
+ when(processor.getParameterLookup()).thenReturn(ParameterLookup.EMPTY);
when(processor.createValidationContext(any(Map.class), any(),
any(ParameterLookup.class), anyBoolean())).thenReturn(validationContext);
return processor;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/MockProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/MockProcessGroup.java
index 77c6e0391c0..fb513cf82a1 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/MockProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/mock/MockProcessGroup.java
@@ -74,9 +74,13 @@ import java.util.function.Predicate;
public class MockProcessGroup implements ProcessGroup {
private final Map<String, ControllerServiceNode> serviceMap = new
HashMap<>();
private final Map<String, ProcessorNode> processorMap = new HashMap<>();
+ private final Map<String, ProcessGroup> processGroupMap = new HashMap<>();
private final Map<String, Port> inputPortMap = new HashMap<>();
private final Map<String, Port> outputPortMap = new HashMap<>();
private final FlowManager flowManager;
+ private final String identifier;
+ private ProcessGroup parent;
+ private ExecutionEngine executionEngine = ExecutionEngine.STANDARD;
private VersionControlInformation versionControlInfo;
private ParameterContext parameterContext;
private String defaultFlowfileExpiration;
@@ -84,7 +88,12 @@ public class MockProcessGroup implements ProcessGroup {
private String defaultBackPressureDataSizeThreshold;
public MockProcessGroup(final FlowManager flowManager) {
+ this(flowManager, "unit test group id");
+ }
+
+ public MockProcessGroup(final FlowManager flowManager, final String
identifier) {
this.flowManager = flowManager;
+ this.identifier = identifier;
}
@Override
@@ -99,7 +108,7 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public ProcessGroup getParent() {
- return null;
+ return parent;
}
@Override
@@ -109,12 +118,12 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public void setParent(final ProcessGroup group) {
-
+ parent = group;
}
@Override
public String getIdentifier() {
- return "unit test group id";
+ return identifier;
}
@Override
@@ -192,6 +201,7 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public void setExecutionEngine(final ExecutionEngine executionEngine) {
+ this.executionEngine = executionEngine;
}
@Override
@@ -335,22 +345,23 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public void addProcessGroup(final ProcessGroup group) {
-
+ group.setParent(this);
+ processGroupMap.put(group.getIdentifier(), group);
}
@Override
public ProcessGroup getProcessGroup(final String id) {
- return null;
+ return processGroupMap.get(id);
}
@Override
public Set<ProcessGroup> getProcessGroups() {
- return null;
+ return new HashSet<>(processGroupMap.values());
}
@Override
public void removeProcessGroup(final ProcessGroup group) {
-
+ processGroupMap.remove(group.getIdentifier());
}
@Override
@@ -527,7 +538,12 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public List<ProcessorNode> findAllProcessors() {
- return new ArrayList<>(processorMap.values());
+ final List<ProcessorNode> processors = new
ArrayList<>(processorMap.values());
+ for (final ProcessGroup processGroup : processGroupMap.values()) {
+ processors.addAll(processGroup.findAllProcessors());
+ }
+
+ return processors;
}
@Override
@@ -875,12 +891,16 @@ public class MockProcessGroup implements ProcessGroup {
@Override
public ExecutionEngine getExecutionEngine() {
- return ExecutionEngine.STANDARD;
+ return executionEngine;
}
@Override
public ExecutionEngine resolveExecutionEngine() {
- return ExecutionEngine.STANDARD;
+ if (executionEngine == ExecutionEngine.INHERITED) {
+ return parent == null ? ExecutionEngine.STANDARD :
parent.resolveExecutionEngine();
+ }
+
+ return executionEngine;
}
@Override
