This is an automated email from the ASF dual-hosted git repository.
rfellows 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 751efb02f03 NIFI-16337: Add support for stopping Process Group sources
(#11678)
751efb02f03 is described below
commit 751efb02f0360f89266d625cbda8e7c3a00465ee
Author: Matt Gilman <[email protected]>
AuthorDate: Tue Sep 15 15:18:43 2026 -0400
NIFI-16337: Add support for stopping Process Group sources (#11678)
* NIFI-16337: Add support for stopping Process Group sources
- add the Stop sources REST endpoint and cluster response handling
- expose resolved execution engines on Process Group payloads
- add Flow Designer actions for stopping source components
- reject STATELESS targets and skip nested STATELESS groups
- add backend and frontend test coverage
* NIFI-16337: Addressing review feedback.
---
.../apache/nifi/web/api/dto/ProcessGroupDTO.java | 12 +
.../nifi/web/api/dto/flow/ProcessGroupFlowDTO.java | 12 +
.../nifi/web/api/entity/ProcessGroupEntity.java | 13 +
.../http/StandardHttpResponseMapper.java | 2 +
.../http/endpoints/StopSourcesEndpointMerger.java | 60 +++++
.../endpoints/StopSourcesEndpointMergerTest.java | 92 +++++++
.../org/apache/nifi/web/NiFiServiceFacade.java | 30 +++
.../apache/nifi/web/StandardNiFiServiceFacade.java | 61 +++++
.../java/org/apache/nifi/web/api/FlowResource.java | 119 +++++++++
.../org/apache/nifi/web/api/dto/DtoFactory.java | 4 +
.../org/apache/nifi/web/api/dto/EntityFactory.java | 1 +
.../nifi/web/StandardNiFiServiceFacadeTest.java | 290 +++++++++++++++++++++
.../org/apache/nifi/web/api/TestFlowResource.java | 251 ++++++++++++++++++
.../apache/nifi/web/api/dto/DtoFactoryTest.java | 106 ++++++++
.../apache/nifi/web/api/dto/EntityFactoryTest.java | 58 +++++
.../service/canvas-context-menu.service.spec.ts | 175 +++++++++++++
.../service/canvas-context-menu.service.ts | 31 +++
.../flow-designer/service/canvas-utils.service.ts | 21 +-
.../pages/flow-designer/service/flow.service.ts | 14 +
.../pages/flow-designer/state/flow/flow.actions.ts | 8 +
.../flow-designer/state/flow/flow.effects.spec.ts | 142 +++++++++-
.../pages/flow-designer/state/flow/flow.effects.ts | 58 +++++
.../pages/flow-designer/state/flow/flow.reducer.ts | 5 +
.../flow-designer/state/flow/flow.selectors.ts | 5 +
.../app/pages/flow-designer/state/flow/index.ts | 14 +
.../tests/system/pg/ClusteredStopSourcesIT.java | 260 ++++++++++++++++++
.../org/apache/nifi/toolkit/client/FlowClient.java | 10 +
.../nifi/toolkit/client/impl/JerseyFlowClient.java | 26 ++
28 files changed, 1875 insertions(+), 5 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 1a830d5420e..17cfa451c51 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
@@ -38,6 +38,7 @@ public class ProcessGroupDTO extends ComponentDTO {
private String defaultBackPressureDataSizeThreshold;
private String logFileSuffix;
private String executionEngine;
+ private String resolvedExecutionEngine;
private Integer maxConcurrentTasks;
private String statelessFlowTimeout;
@@ -395,6 +396,17 @@ public class ProcessGroupDTO extends ComponentDTO {
this.executionEngine = executionEngine;
}
+ @Schema(description = "The Execution Engine that will actually run this
Process Group after resolving INHERITED. Never INHERITED; a root group
configured as INHERITED resolves to STANDARD.",
+ allowableValues = {"STATELESS", "STANDARD"},
+ accessMode = Schema.AccessMode.READ_ONLY)
+ public String getResolvedExecutionEngine() {
+ return resolvedExecutionEngine;
+ }
+
+ public void setResolvedExecutionEngine(final String
resolvedExecutionEngine) {
+ this.resolvedExecutionEngine = resolvedExecutionEngine;
+ }
+
@Schema(description = "If the Process Group is configured to run in using
the Stateless Engine, represents the current state. Otherwise, will be
STOPPED.",
allowableValues = {"STOPPED", "RUNNING"})
public String getStatelessGroupScheduledState() {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/flow/ProcessGroupFlowDTO.java
b/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/flow/ProcessGroupFlowDTO.java
index f2bdaea9db9..a430d88c9d2 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/flow/ProcessGroupFlowDTO.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/flow/ProcessGroupFlowDTO.java
@@ -38,6 +38,7 @@ public class ProcessGroupFlowDTO {
private FlowBreadcrumbEntity breadcrumb;
private FlowDTO flow;
private Date lastRefreshed;
+ private String resolvedExecutionEngine;
/**
* @return contents of this process group. This field will be populated if
the request is marked verbose
@@ -130,4 +131,15 @@ public class ProcessGroupFlowDTO {
public void setParameterContext(ParameterContextReferenceEntity
parameterContext) {
this.parameterContext = parameterContext;
}
+
+ @Schema(description = "The Execution Engine that will actually run this
Process Group after resolving INHERITED. Never INHERITED; a root group
configured as INHERITED resolves to STANDARD.",
+ allowableValues = {"STATELESS", "STANDARD"},
+ accessMode = Schema.AccessMode.READ_ONLY)
+ public String getResolvedExecutionEngine() {
+ return resolvedExecutionEngine;
+ }
+
+ public void setResolvedExecutionEngine(final String
resolvedExecutionEngine) {
+ this.resolvedExecutionEngine = resolvedExecutionEngine;
+ }
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ProcessGroupEntity.java
b/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ProcessGroupEntity.java
index 70fddd5eaf8..21eb63d51ff 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ProcessGroupEntity.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ProcessGroupEntity.java
@@ -56,6 +56,7 @@ public class ProcessGroupEntity extends ComponentEntity
implements Permissible<P
private ParameterContextReferenceEntity parameterContext;
private String processGroupUpdateStrategy;
+ private String resolvedExecutionEngine;
/**
* The ProcessGroupDTO that is being serialized.
@@ -329,4 +330,16 @@ public class ProcessGroupEntity extends ComponentEntity
implements Permissible<P
public void setProcessGroupUpdateStrategy(String
processGroupUpdateStrategy) {
this.processGroupUpdateStrategy = processGroupUpdateStrategy;
}
+
+ @Schema(description = "The Execution Engine that will actually run this
Process Group after resolving INHERITED. Never INHERITED; a root group
configured as INHERITED resolves to STANDARD. "
+ + "Promoted onto the entity so it is available when the user
cannot read the Process Group.",
+ allowableValues = {"STATELESS", "STANDARD"},
+ accessMode = Schema.AccessMode.READ_ONLY)
+ public String getResolvedExecutionEngine() {
+ return resolvedExecutionEngine;
+ }
+
+ public void setResolvedExecutionEngine(final String
resolvedExecutionEngine) {
+ this.resolvedExecutionEngine = resolvedExecutionEngine;
+ }
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
index 8fdd9a498e0..a760c045bdf 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
@@ -102,6 +102,7 @@ import
org.apache.nifi.cluster.coordination.http.endpoints.RuleViolationEndpoint
import
org.apache.nifi.cluster.coordination.http.endpoints.RuntimeManifestEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.SearchUsersEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.StatusHistoryEndpointMerger;
+import
org.apache.nifi.cluster.coordination.http.endpoints.StopSourcesEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.SystemDiagnosticsEndpointMerger;
import org.apache.nifi.cluster.coordination.http.endpoints.UserEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.UserGroupEndpointMerger;
@@ -191,6 +192,7 @@ public class StandardHttpResponseMapper implements
HttpResponseMapper {
endpointMergers.add(new SystemDiagnosticsEndpointMerger());
endpointMergers.add(new CountersEndpointMerger());
endpointMergers.add(new FlowMerger());
+ endpointMergers.add(new StopSourcesEndpointMerger());
endpointMergers.add(new ProcessorTypesEndpointMerger());
endpointMergers.add(new ControllerServiceTypesEndpointMerger());
endpointMergers.add(new ReportingTaskTypesEndpointMerger());
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMerger.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMerger.java
new file mode 100644
index 00000000000..16b2daafe85
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMerger.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.dto.RevisionDTO;
+import org.apache.nifi.web.api.entity.ScheduleComponentsEntity;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+public class StopSourcesEndpointMerger extends
AbstractSingleEntityEndpoint<ScheduleComponentsEntity> {
+ public static final Pattern STOP_SOURCES_URI_PATTERN =
Pattern.compile("/nifi-api/flow/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/sources");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "PUT".equalsIgnoreCase(method) &&
STOP_SOURCES_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ protected Class<ScheduleComponentsEntity> getEntityClass() {
+ return ScheduleComponentsEntity.class;
+ }
+
+ @Override
+ protected void mergeResponses(final ScheduleComponentsEntity clientEntity,
final Map<NodeIdentifier, ScheduleComponentsEntity> entityMap,
+ final Set<NodeResponse> successfulResponses,
final Set<NodeResponse> problematicResponses) {
+ if (clientEntity.getComponents() == null) {
+ clientEntity.setComponents(new HashMap<>());
+ }
+
+ for (final ScheduleComponentsEntity nodeEntity : entityMap.values()) {
+ if (nodeEntity.getComponents() == null) {
+ continue;
+ }
+
+ for (final Map.Entry<String, RevisionDTO> entry :
nodeEntity.getComponents().entrySet()) {
+ clientEntity.getComponents().putIfAbsent(entry.getKey(),
entry.getValue());
+ }
+ }
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMergerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMergerTest.java
new file mode 100644
index 00000000000..3f0f48b3b6f
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StopSourcesEndpointMergerTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.web.api.dto.RevisionDTO;
+import org.apache.nifi.web.api.entity.ScheduleComponentsEntity;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class StopSourcesEndpointMergerTest {
+
+ private static final String GROUP_ID =
"12345678-1234-1234-1234-123456789012";
+
+ @Test
+ public void testCanHandle() {
+ final StopSourcesEndpointMerger merger = new
StopSourcesEndpointMerger();
+
+
assertTrue(merger.canHandle(URI.create("/nifi-api/flow/process-groups/" +
GROUP_ID + "/sources"), "PUT"));
+
assertTrue(merger.canHandle(URI.create("/nifi-api/flow/process-groups/root/sources"),
"PUT"));
+
assertTrue(merger.canHandle(URI.create("/nifi-api/flow/process-groups/" +
GROUP_ID + "/sources"), "put"));
+
+
assertFalse(merger.canHandle(URI.create("/nifi-api/flow/process-groups/" +
GROUP_ID + "/sources"), "GET"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/flow/process-groups/" +
GROUP_ID), "PUT"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/flow/process-groups/" +
GROUP_ID), "GET"));
+ assertFalse(merger.canHandle(URI.create("/nifi-api/process-groups/" +
GROUP_ID + "/sources"), "PUT"));
+ }
+
+ @Test
+ public void testMergeResponsesUnionsComponentsAndKeepsClientRevision() {
+ final StopSourcesEndpointMerger merger = new
StopSourcesEndpointMerger();
+
+ final RevisionDTO clientRevision = revision(1L, "client");
+ final RevisionDTO nodeRevision = revision(2L, "node-2");
+ final RevisionDTO extraRevision = revision(3L, "node-2-extra");
+
+ final ScheduleComponentsEntity clientEntity = new
ScheduleComponentsEntity();
+ clientEntity.setId(GROUP_ID);
+ clientEntity.setState("STOPPED");
+ clientEntity.setComponents(new HashMap<>(Map.of("source-1",
clientRevision)));
+
+ final NodeIdentifier node1 = new NodeIdentifier("node1", "localhost",
8080, "localhost", 8081, "localhost", 8082, 8083, false);
+ final NodeIdentifier node2 = new NodeIdentifier("node2", "localhost",
8090, "localhost", 8091, "localhost", 8092, 8093, false);
+
+ final ScheduleComponentsEntity node1Entity = new
ScheduleComponentsEntity();
+ node1Entity.setId(GROUP_ID);
+ node1Entity.setState("STOPPED");
+ node1Entity.setComponents(Map.of("source-1", clientRevision));
+
+ final ScheduleComponentsEntity node2Entity = new
ScheduleComponentsEntity();
+ node2Entity.setId(GROUP_ID);
+ node2Entity.setState("STOPPED");
+ node2Entity.setComponents(Map.of("source-1", nodeRevision, "source-2",
extraRevision));
+
+ merger.mergeResponses(clientEntity, Map.of(node1, node1Entity, node2,
node2Entity), null, null);
+
+ assertEquals(GROUP_ID, clientEntity.getId());
+ assertEquals("STOPPED", clientEntity.getState());
+ assertEquals(2, clientEntity.getComponents().size());
+ assertSame(clientRevision,
clientEntity.getComponents().get("source-1"));
+ assertSame(extraRevision,
clientEntity.getComponents().get("source-2"));
+ }
+
+ private static RevisionDTO revision(final long version, final String
clientId) {
+ final RevisionDTO revision = new RevisionDTO();
+ revision.setVersion(version);
+ revision.setClientId(clientId);
+ return revision;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
index 2313b34f977..2f9b8538d25 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
@@ -387,6 +387,36 @@ public interface NiFiServiceFacade {
*/
Set<Revision> getRevisionsFromGroup(String groupId, Function<ProcessGroup,
Set<String>> getComponents);
+ /**
+ * Identifies source component identifiers in the given Process Group and
its Standard-engine descendants.
+ * Sources are processors with no non-loop incoming connection, Remote
Process Group output ports,
+ * and public input ports with no non-loop incoming connection. Components
owned by Process Groups
+ * that resolve to the Stateless Execution Engine are excluded.
+ *
+ * @param group the process group to search
+ * @return identifiers of source components
+ */
+ Set<String> findSourceComponentIds(ProcessGroup group);
+
+ /**
+ * Verifies that source components can be stopped in the specified Process
Group.
+ *
+ * @param groupId process group identifier
+ * @throws IllegalStateException when the Process Group resolves to the
Stateless Execution Engine
+ */
+ void verifyStopSources(String groupId);
+
+ /**
+ * Verifies that the supplied component identifiers exactly match the
source components currently identified
+ * in the specified Process Group. This ensures that every cluster node
operates on the same source components.
+ *
+ * @param groupId process group identifier
+ * @param componentIds source component identifiers
+ * @throws IllegalStateException when the Process Group resolves to the
Stateless Execution Engine or the supplied
+ * component identifiers do not match the source components currently
identified
+ */
+ void verifyStopSources(String groupId, Set<String> componentIds);
+
/**
* Gets the revisions from the specified snippet.
*
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 b178d39d676..8e2236e34ce 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
@@ -234,6 +234,7 @@ import
org.apache.nifi.registry.flow.mapping.InstantiatedVersionedProcessGroup;
import
org.apache.nifi.registry.flow.mapping.InstantiatedVersionedRemoteGroupPort;
import org.apache.nifi.registry.flow.mapping.VersionedComponentFlowMapper;
import org.apache.nifi.registry.flow.mapping.VersionedComponentStateLookup;
+import org.apache.nifi.remote.PublicPort;
import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinQuery;
@@ -242,6 +243,7 @@ import org.apache.nifi.reporting.ComponentType;
import org.apache.nifi.reporting.ReportingTask;
import org.apache.nifi.reporting.VerifiableReportingTask;
import org.apache.nifi.util.BundleUtils;
+import org.apache.nifi.util.Connectables;
import org.apache.nifi.util.FlowDifferenceFilters;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.StringUtils;
@@ -595,6 +597,65 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
return componentIds.stream().map(id ->
revisionManager.getRevision(id)).collect(Collectors.toSet());
}
+ @Override
+ public Set<String> findSourceComponentIds(final ProcessGroup group) {
+ final Set<String> sourceIds = new LinkedHashSet<>();
+
+ for (final ProcessorNode processor : group.findAllProcessors()) {
+ if (processor.getProcessGroup().resolveExecutionEngine() ==
ExecutionEngine.STANDARD
+ && ProcessGroup.STOP_PROCESSORS_FILTER.test(processor)
+ && !Connectables.hasNonLoopConnection(processor)) {
+ sourceIds.add(processor.getIdentifier());
+ }
+ }
+
+ for (final RemoteProcessGroup remoteProcessGroup :
group.findAllRemoteProcessGroups()) {
+ if (remoteProcessGroup.getProcessGroup().resolveExecutionEngine()
== ExecutionEngine.STATELESS) {
+ continue;
+ }
+
+ for (final RemoteGroupPort remotePort :
remoteProcessGroup.getOutputPorts()) {
+ if (ProcessGroup.STOP_PORTS_FILTER.test(remotePort)) {
+ sourceIds.add(remotePort.getIdentifier());
+ }
+ }
+ }
+
+ for (final Port port : group.findAllInputPorts()) {
+ if (port.getProcessGroup().resolveExecutionEngine() ==
ExecutionEngine.STANDARD
+ && port instanceof PublicPort
+ && ProcessGroup.STOP_PORTS_FILTER.test(port)
+ && !Connectables.hasNonLoopConnection(port)) {
+ sourceIds.add(port.getIdentifier());
+ }
+ }
+
+ return sourceIds;
+ }
+
+ @Override
+ public void verifyStopSources(final String groupId) {
+ final ProcessGroup group = processGroupDAO.getProcessGroup(groupId);
+ verifyStopSources(group);
+ }
+
+ @Override
+ public void verifyStopSources(final String groupId, final Set<String>
componentIds) {
+ final ProcessGroup group = processGroupDAO.getProcessGroup(groupId);
+ verifyStopSources(group);
+
+ final Set<String> sourceComponentIds = findSourceComponentIds(group);
+ if (!sourceComponentIds.equals(componentIds)) {
+ throw new IllegalStateException("Source components changed while
processing the request; refresh and retry");
+ }
+ }
+
+ private static void verifyStopSources(final ProcessGroup group) {
+ if (group.resolveExecutionEngine() == ExecutionEngine.STATELESS) {
+ throw new IllegalStateException("Cannot stop sources in a Process
Group that resolves to the Stateless Execution Engine");
+ }
+ }
+
@Override
public Set<Revision> getRevisionsFromSnippet(final String snippetId) {
final Snippet snippet = snippetDAO.getSnippet(snippetId);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowResource.java
index 753c6c66a97..a77f79fdba7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowResource.java
@@ -1217,6 +1217,125 @@ public class FlowResource extends ApplicationResource {
);
}
+ /**
+ * Stops source components in the specified process group and its
Standard-engine descendants. Sources are
+ * processors with no non-loop incoming connection, Remote Process Group
output ports, and public input ports
+ * with no non-loop incoming connection. The operation is rejected when
the specified group resolves to the
+ * Stateless Execution Engine.
+ *
+ * @param id The id of the process group.
+ * @param requestScheduleComponentsEntity A scheduleComponentsEntity with
state STOPPED.
+ * @return A scheduleComponentsEntity identifying the components that were
stopped.
+ */
+ @PUT
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("process-groups/{id}/sources")
+ @Operation(
+ summary = "Stop source components in the specified Process Group.",
+ description = "All source components included in the operation
must be authorized. If any source is unauthorized, no components are stopped.",
+ responses = {
+ @ApiResponse(responseCode = "200", content =
@Content(schema = @Schema(implementation = ScheduleComponentsEntity.class))),
+ @ApiResponse(responseCode = "400", description = "NiFi was
unable to complete the request because it was invalid. The request should not
be retried without modification."),
+ @ApiResponse(responseCode = "401", description = "Client
could not be authenticated."),
+ @ApiResponse(responseCode = "403", description = "Client
is not authorized to make this request."),
+ @ApiResponse(responseCode = "404", description = "The
specified resource could not be found."),
+ @ApiResponse(responseCode = "409", description = "The
request was valid but NiFi was not in the appropriate state to process it.")
+ },
+ security = {
+ @SecurityRequirement(name = "Read - /flow"),
+ @SecurityRequirement(name = "Write -
/{component-type}/{uuid} or /operation/{component-type}/{uuid} - For every
source component being stopped")
+ }
+ )
+ public Response stopSources(
+ @Parameter(
+ description = "The process group id.",
+ required = true
+ )
+ @PathParam("id") final String id,
+ @Parameter(
+ description = "The request to stop sources. If the
components in the request are not specified, all source components are
identified.",
+ required = true
+ ) final ScheduleComponentsEntity requestScheduleComponentsEntity) {
+
+ if (requestScheduleComponentsEntity == null) {
+ throw new IllegalArgumentException("Schedule Component must be
specified.");
+ }
+
+ if (!id.equals(requestScheduleComponentsEntity.getId())) {
+ throw new IllegalArgumentException(String.format("The process
group id (%s) in the request body does "
+ + "not equal the process group id of the requested
resource (%s).", requestScheduleComponentsEntity.getId(), id));
+ }
+
+ if
(!ScheduledState.STOPPED.name().equals(requestScheduleComponentsEntity.getState()))
{
+ throw new IllegalArgumentException("The scheduled state must be
STOPPED.");
+ }
+
+ authorizeFlow();
+ serviceFacade.verifyStopSources(id);
+
+ if (requestScheduleComponentsEntity.getComponents() == null) {
+ final Set<Revision> revisions =
serviceFacade.getRevisionsFromGroup(id, serviceFacade::findSourceComponentIds);
+ final Map<String, RevisionDTO> componentsToStop = new HashMap<>();
+ for (final Revision revision : revisions) {
+ final RevisionDTO dto = new RevisionDTO();
+ dto.setClientId(revision.getClientId());
+ dto.setVersion(revision.getVersion());
+ componentsToStop.put(revision.getComponentId(), dto);
+ }
+
+ requestScheduleComponentsEntity.setComponents(componentsToStop);
+ }
+
+ final Map<String, RevisionDTO> requestComponentsToStop =
requestScheduleComponentsEntity.getComponents();
+ final Set<String> requestComponentIds =
Set.copyOf(requestComponentsToStop.keySet());
+
+ serviceFacade.authorizeAccess(lookup -> {
+ for (final String componentId : requestComponentIds) {
+ final Authorizable connectable =
lookup.getLocalConnectable(componentId);
+ OperationAuthorizable.authorizeOperation(connectable,
authorizer, NiFiUserUtils.getNiFiUser());
+ }
+ });
+
+ serviceFacade.verifyStopSources(id, requestComponentIds);
+
+ if (isReplicateRequest()) {
+ return replicate(HttpMethod.PUT, requestScheduleComponentsEntity);
+ } else if (isDisconnectedFromCluster()) {
+
verifyDisconnectedNodeModification(requestScheduleComponentsEntity.isDisconnectedNodeAcknowledged());
+ }
+
+ final Map<String, Revision> requestComponentRevisions =
+
requestComponentsToStop.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
e -> getRevision(e.getValue(), e.getKey())));
+ final Set<Revision> requestRevisions = new
HashSet<>(requestComponentRevisions.values());
+
+ return withWriteLock(
+ serviceFacade,
+ requestScheduleComponentsEntity,
+ requestRevisions,
+ lookup -> {
+ authorizeFlow();
+
+ requestComponentsToStop.keySet().forEach(componentId -> {
+ final Authorizable connectable =
lookup.getLocalConnectable(componentId);
+ OperationAuthorizable.authorizeOperation(connectable,
authorizer, NiFiUserUtils.getNiFiUser());
+ });
+ },
+ () -> {
+ serviceFacade.verifyStopSources(id, requestComponentIds);
+ serviceFacade.verifyScheduleComponents(id,
ScheduledState.STOPPED, requestComponentIds);
+ },
+ (revisions, scheduleComponentsEntity) -> {
+ final Map<String, RevisionDTO> componentsToStop =
scheduleComponentsEntity.getComponents();
+ final Map<String, Revision> componentRevisions =
+
componentsToStop.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,
e -> getRevision(e.getValue(), e.getKey())));
+ final ScheduleComponentsEntity entity =
serviceFacade.scheduleComponents(id, ScheduledState.STOPPED,
componentRevisions);
+ entity.setComponents(componentsToStop);
+ return generateOkResponse(entity).build();
+ }
+ );
+ }
+
@PUT
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 46e8ab35d7a..bc24e38ecda 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -2568,6 +2568,8 @@ public final class DtoFactory {
dto.setParentGroupId(parent.getIdentifier());
}
+ dto.setResolvedExecutionEngine(group.resolveExecutionEngine().name());
+
final ParameterContext parameterContext = group.getParameterContext();
if (parameterContext != null) {
dto.setParameterContext(entityFactory.createParameterReferenceEntity(createParameterContextReference(parameterContext),
createPermissionsDto(parameterContext)));
@@ -2834,6 +2836,7 @@ public final class DtoFactory {
dto.setLogFileSuffix(group.getLogFileSuffix());
dto.setStatelessGroupScheduledState(group.getStatelessScheduledState().name());
dto.setExecutionEngine(group.getExecutionEngine().name());
+ dto.setResolvedExecutionEngine(group.resolveExecutionEngine().name());
dto.setMaxConcurrentTasks(group.getMaxConcurrentTasks());
dto.setStatelessFlowTimeout(group.getStatelessFlowTimeout());
@@ -4825,6 +4828,7 @@ public final class DtoFactory {
copy.setDefaultBackPressureDataSizeThreshold(original.getDefaultBackPressureDataSizeThreshold());
copy.setLogFileSuffix(original.getLogFileSuffix());
copy.setExecutionEngine(original.getExecutionEngine());
+ copy.setResolvedExecutionEngine(original.getResolvedExecutionEngine());
copy.setMaxConcurrentTasks(original.getMaxConcurrentTasks());
copy.setStatelessFlowTimeout(original.getStatelessFlowTimeout());
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/EntityFactory.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/EntityFactory.java
index be52686de48..f3733142082 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/EntityFactory.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/EntityFactory.java
@@ -320,6 +320,7 @@ public final class EntityFactory {
entity.setStaleCount(dto.getStaleCount());
entity.setLocallyModifiedAndStaleCount(dto.getLocallyModifiedAndStaleCount());
entity.setSyncFailureCount(dto.getSyncFailureCount());
+
entity.setResolvedExecutionEngine(dto.getResolvedExecutionEngine());
final ParameterContextReferenceEntity parameterContextReference =
dto.getParameterContext();
if (parameterContextReference != null) {
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 079f8cf8629..47f729e1763 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
@@ -56,6 +56,9 @@ import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.components.validation.ValidationStatus;
+import org.apache.nifi.connectable.Connectable;
+import org.apache.nifi.connectable.Connection;
+import org.apache.nifi.connectable.Port;
import org.apache.nifi.controller.ClusterTopologyProvider;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.Counter;
@@ -114,6 +117,8 @@ import
org.apache.nifi.registry.flow.diff.StaticDifferenceDescriptor;
import org.apache.nifi.registry.flow.mapping.FlowMappingOptions;
import org.apache.nifi.registry.flow.mapping.InstantiatedVersionedProcessGroup;
import org.apache.nifi.registry.flow.mapping.VersionedComponentFlowMapper;
+import org.apache.nifi.remote.PublicPort;
+import org.apache.nifi.remote.RemoteGroupPort;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinFactory;
import org.apache.nifi.reporting.BulletinQuery;
@@ -3260,6 +3265,291 @@ public class StandardNiFiServiceFacadeTest {
verify(assetManager).deleteAsset(ASSET_ID);
}
+ @Test
+ public void
testFindSourceComponentIdsIncludesProcessorWithNoIncomingConnections() {
+ final ProcessorNode processor = sourceProcessor("source-processor");
+ final ProcessGroup group = sourceProcessGroup(List.of(processor),
List.of(), List.of());
+
+ assertEquals(Set.of("source-processor"),
serviceFacade.findSourceComponentIds(group));
+ }
+
+ @Test
+ public void testFindSourceComponentIdsIncludesProcessorWithSelfLoop() {
+ final ProcessorNode processor = sourceProcessor("self-loop-processor");
+ final Connection selfLoop = connectionFrom(processor);
+ when(processor.getIncomingConnections()).thenReturn(List.of(selfLoop));
+ final ProcessGroup group = sourceProcessGroup(List.of(processor),
List.of(), List.of());
+
+ assertEquals(Set.of("self-loop-processor"),
serviceFacade.findSourceComponentIds(group));
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsExcludesProcessorWithIncomingConnectionFromOtherComponent()
{
+ final ProcessorNode upstream = sourceProcessor("upstream");
+ final ProcessorNode downstream = sourceProcessor("downstream");
+ final Connection incomingConnection = connectionFrom(upstream);
+
when(downstream.getIncomingConnections()).thenReturn(List.of(incomingConnection));
+ final ProcessGroup group = sourceProcessGroup(List.of(upstream,
downstream), List.of(), List.of());
+
+ assertEquals(Set.of("upstream"),
serviceFacade.findSourceComponentIds(group));
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsIncludesNestedProcessGroupProcessors() {
+ final ProcessorNode parentSource = sourceProcessor("parent-source");
+ final ProcessorNode nestedSource = sourceProcessor("nested-source");
+ final ProcessGroup group = sourceProcessGroup(List.of(parentSource,
nestedSource), List.of(), List.of());
+
+ assertEquals(Set.of("parent-source", "nested-source"),
serviceFacade.findSourceComponentIds(group));
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsExcludesSourcesInStatelessProcessGroups() {
+ final ProcessorNode statelessProcessor =
sourceProcessor("stateless-processor");
+ final RemoteGroupPort statelessRemoteOutput =
remotePort("stateless-remote-output");
+ final RemoteProcessGroup statelessRemoteProcessGroup =
mock(RemoteProcessGroup.class);
+
when(statelessRemoteProcessGroup.getOutputPorts()).thenReturn(Set.of(statelessRemoteOutput));
+ final PublicPort statelessPublicInput =
sourcePublicPort("stateless-public-input");
+ final ProcessGroup group = sourceProcessGroup(
+ List.of(statelessProcessor),
+ List.of(statelessRemoteProcessGroup),
+ List.of(statelessPublicInput)
+ );
+ final ProcessGroup statelessGroup = mock(ProcessGroup.class);
+
when(statelessGroup.resolveExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+ when(statelessProcessor.getProcessGroup()).thenReturn(statelessGroup);
+
when(statelessRemoteProcessGroup.getProcessGroup()).thenReturn(statelessGroup);
+
when(statelessPublicInput.getProcessGroup()).thenReturn(statelessGroup);
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void testVerifyStopSourcesRejectsStatelessProcessGroup() {
+ final ProcessGroup group = mock(ProcessGroup.class);
+
when(group.resolveExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+
when(processGroupDAO.getProcessGroup("stateless-group")).thenReturn(group);
+
+ assertThrows(IllegalStateException.class, () ->
serviceFacade.verifyStopSources("stateless-group"));
+ }
+
+ @Test
+ public void testVerifyStopSourcesAllowsStandardProcessGroup() {
+ final ProcessGroup group = mock(ProcessGroup.class);
+
when(group.resolveExecutionEngine()).thenReturn(ExecutionEngine.STANDARD);
+
when(processGroupDAO.getProcessGroup("standard-group")).thenReturn(group);
+
+ serviceFacade.verifyStopSources("standard-group");
+ }
+
+ @Test
+ public void testVerifyStopSourcesAllowsMatchingSourceComponents() {
+ final ProcessorNode sourceProcessor =
sourceProcessor("source-processor");
+ final ProcessGroup group =
sourceProcessGroup(List.of(sourceProcessor), List.of(), List.of());
+
when(processGroupDAO.getProcessGroup("standard-group")).thenReturn(group);
+
+ serviceFacade.verifyStopSources("standard-group",
Set.of("source-processor"));
+ }
+
+ @Test
+ public void testVerifyStopSourcesRejectsMissingSourceComponent() {
+ final ProcessorNode sourceProcessor =
sourceProcessor("source-processor");
+ final ProcessGroup group =
sourceProcessGroup(List.of(sourceProcessor), List.of(), List.of());
+
when(processGroupDAO.getProcessGroup("standard-group")).thenReturn(group);
+
+ assertThrows(IllegalStateException.class, () ->
serviceFacade.verifyStopSources("standard-group", Set.of()));
+ }
+
+ @Test
+ public void testVerifyStopSourcesRejectsNonSourceComponent() {
+ final ProcessorNode sourceProcessor =
sourceProcessor("source-processor");
+ final ProcessorNode downstreamProcessor =
sourceProcessor("downstream-processor");
+ final Connection incomingConnection = connectionFrom(sourceProcessor);
+
when(downstreamProcessor.getIncomingConnections()).thenReturn(List.of(incomingConnection));
+ final ProcessGroup group = sourceProcessGroup(List.of(sourceProcessor,
downstreamProcessor), List.of(), List.of());
+
when(processGroupDAO.getProcessGroup("standard-group")).thenReturn(group);
+
+ assertThrows(IllegalStateException.class,
+ () -> serviceFacade.verifyStopSources("standard-group",
Set.of("source-processor", "downstream-processor")));
+ }
+
+ @Test
+ public void testVerifyStopSourcesRejectsComponentInStatelessDescendant() {
+ final ProcessorNode statelessProcessor =
sourceProcessor("stateless-processor");
+ final ProcessGroup group =
sourceProcessGroup(List.of(statelessProcessor), List.of(), List.of());
+ final ProcessGroup statelessGroup = mock(ProcessGroup.class);
+
when(statelessGroup.resolveExecutionEngine()).thenReturn(ExecutionEngine.STATELESS);
+ when(statelessProcessor.getProcessGroup()).thenReturn(statelessGroup);
+
when(processGroupDAO.getProcessGroup("standard-group")).thenReturn(group);
+
+ assertThrows(IllegalStateException.class,
+ () -> serviceFacade.verifyStopSources("standard-group",
Set.of("stateless-processor")));
+ }
+
+ @Test
+ public void testFindSourceComponentIdsExcludesNonRunningSourceProcessor() {
+ final ProcessorNode nonRunningSource =
sourceProcessor("non-running-source", false);
+ final ProcessGroup group =
sourceProcessGroup(List.of(nonRunningSource), List.of(), List.of());
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsExcludesNonTransmittingRemoteOutputPort() {
+ final RemoteGroupPort remoteOutput = remotePort("remote-output",
ScheduledState.STOPPED);
+ final RemoteProcessGroup remoteProcessGroup =
mock(RemoteProcessGroup.class);
+
when(remoteProcessGroup.getOutputPorts()).thenReturn(Set.of(remoteOutput));
+ when(remoteProcessGroup.getInputPorts()).thenReturn(Set.of());
+ final ProcessGroup group = sourceProcessGroup(List.of(),
List.of(remoteProcessGroup), List.of());
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void testFindSourceComponentIdsExcludesStoppedPublicInputPort() {
+ final PublicPort publicInput = sourcePublicPort("public-input",
ScheduledState.STOPPED);
+ final ProcessGroup group = sourceProcessGroup(List.of(), List.of(),
List.of(publicInput));
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsIncludesRemoteOutputPortAndExcludesRemoteInputPort() {
+ final RemoteGroupPort remoteOutput = remotePort("remote-output");
+ final RemoteGroupPort remoteInput = remotePort("remote-input");
+ final RemoteProcessGroup remoteProcessGroup =
mock(RemoteProcessGroup.class);
+ when(remoteProcessGroup.getIdentifier()).thenReturn("rpg");
+
when(remoteProcessGroup.getOutputPorts()).thenReturn(Set.of(remoteOutput));
+
when(remoteProcessGroup.getInputPorts()).thenReturn(Set.of(remoteInput));
+ final ProcessGroup group = sourceProcessGroup(List.of(),
List.of(remoteProcessGroup), List.of());
+
+ final Set<String> sourceIds =
serviceFacade.findSourceComponentIds(group);
+ assertEquals(Set.of("remote-output"), sourceIds);
+ assertFalse(sourceIds.contains("rpg"));
+ assertFalse(sourceIds.contains("remote-input"));
+ }
+
+ @Test
+ public void testFindSourceComponentIdsExcludesLocalPorts() {
+ final Port localInput = sourcePort("local-input");
+ final Port localOutput = sourcePort("local-output");
+ final ProcessGroup group = sourceProcessGroup(List.of(), List.of(),
List.of(localInput));
+ when(group.findAllOutputPorts()).thenReturn(List.of(localOutput));
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsIncludesPublicInputPortWithNoIncomingConnections() {
+ final PublicPort publicInput = sourcePublicPort("public-input");
+ final ProcessGroup group = sourceProcessGroup(List.of(), List.of(),
List.of(publicInput));
+
+ assertEquals(Set.of("public-input"),
serviceFacade.findSourceComponentIds(group));
+ }
+
+ @Test
+ public void
testFindSourceComponentIdsExcludesPublicInputPortWithIncomingConnectionFromParent()
{
+ final Connectable parentOutput = sourceProcessor("parent-output");
+ final PublicPort publicInput = sourcePublicPort("nested-public-input");
+ final Connection incomingConnection = connectionFrom(parentOutput);
+
when(publicInput.getIncomingConnections()).thenReturn(List.of(incomingConnection));
+ final ProcessGroup group = sourceProcessGroup(List.of(), List.of(),
List.of(publicInput));
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void testFindSourceComponentIdsExcludesPublicOutputPort() {
+ final PublicPort publicOutput = sourcePublicPort("public-output");
+ final ProcessGroup group = sourceProcessGroup(List.of(), List.of(),
List.of());
+ when(group.findAllOutputPorts()).thenReturn(List.of(publicOutput));
+
+ assertTrue(serviceFacade.findSourceComponentIds(group).isEmpty());
+ }
+
+ @Test
+ public void testFindSourceComponentIdsMixOfSourcesAndNonSources() {
+ final ProcessorNode sourceProcessor =
sourceProcessor("source-processor");
+ final ProcessorNode downstream = sourceProcessor("downstream");
+ final Connection incomingConnection = connectionFrom(sourceProcessor);
+
when(downstream.getIncomingConnections()).thenReturn(List.of(incomingConnection));
+
+ final RemoteGroupPort remoteOutput = remotePort("remote-output");
+ final RemoteGroupPort remoteInput = remotePort("remote-input");
+ final RemoteProcessGroup remoteProcessGroup =
mock(RemoteProcessGroup.class);
+ when(remoteProcessGroup.getIdentifier()).thenReturn("rpg");
+
when(remoteProcessGroup.getOutputPorts()).thenReturn(Set.of(remoteOutput));
+
when(remoteProcessGroup.getInputPorts()).thenReturn(Set.of(remoteInput));
+
+ final PublicPort publicInput = sourcePublicPort("public-input");
+ final Port localInput = sourcePort("local-input");
+
+ final ProcessGroup group = sourceProcessGroup(List.of(sourceProcessor,
downstream), List.of(remoteProcessGroup), List.of(publicInput, localInput));
+
+ assertEquals(Set.of("source-processor", "remote-output",
"public-input"), serviceFacade.findSourceComponentIds(group));
+ }
+
+ private static ProcessGroup sourceProcessGroup(final List<ProcessorNode>
processors, final List<RemoteProcessGroup> remoteProcessGroups, final
List<Port> inputPorts) {
+ final ProcessGroup group = mock(ProcessGroup.class);
+
when(group.resolveExecutionEngine()).thenReturn(ExecutionEngine.STANDARD);
+ when(group.findAllProcessors()).thenReturn(processors);
+
when(group.findAllRemoteProcessGroups()).thenReturn(remoteProcessGroups);
+ when(group.findAllInputPorts()).thenReturn(inputPorts);
+ when(group.findAllOutputPorts()).thenReturn(Collections.emptyList());
+ processors.forEach(processor ->
when(processor.getProcessGroup()).thenReturn(group));
+ remoteProcessGroups.forEach(remoteProcessGroup ->
when(remoteProcessGroup.getProcessGroup()).thenReturn(group));
+ inputPorts.forEach(inputPort ->
when(inputPort.getProcessGroup()).thenReturn(group));
+ return group;
+ }
+
+ private static ProcessorNode sourceProcessor(final String identifier) {
+ return sourceProcessor(identifier, true);
+ }
+
+ private static ProcessorNode sourceProcessor(final String identifier,
final boolean running) {
+ final ProcessorNode processor = mock(ProcessorNode.class);
+ when(processor.getIdentifier()).thenReturn(identifier);
+
when(processor.getIncomingConnections()).thenReturn(Collections.emptyList());
+ when(processor.isRunning()).thenReturn(running);
+ return processor;
+ }
+
+ private static Port sourcePort(final String identifier) {
+ final Port port = mock(Port.class);
+ when(port.getIdentifier()).thenReturn(identifier);
+
when(port.getIncomingConnections()).thenReturn(Collections.emptyList());
+ return port;
+ }
+
+ private static PublicPort sourcePublicPort(final String identifier) {
+ return sourcePublicPort(identifier, ScheduledState.RUNNING);
+ }
+
+ private static PublicPort sourcePublicPort(final String identifier, final
ScheduledState scheduledState) {
+ final PublicPort port = mock(PublicPort.class);
+ when(port.getIdentifier()).thenReturn(identifier);
+
when(port.getIncomingConnections()).thenReturn(Collections.emptyList());
+ when(port.getScheduledState()).thenReturn(scheduledState);
+ return port;
+ }
+
+ private static RemoteGroupPort remotePort(final String identifier) {
+ return remotePort(identifier, ScheduledState.RUNNING);
+ }
+
+ private static RemoteGroupPort remotePort(final String identifier, final
ScheduledState scheduledState) {
+ final RemoteGroupPort port = mock(RemoteGroupPort.class);
+ when(port.getIdentifier()).thenReturn(identifier);
+ when(port.getScheduledState()).thenReturn(scheduledState);
+ return port;
+ }
+
+ private static Connection connectionFrom(final Connectable source) {
+ final Connection connection = mock(Connection.class);
+ when(connection.getSource()).thenReturn(source);
+ return connection;
+ }
+
private Asset createAsset(final String assetId, final String ownerId) {
final Asset asset = mock(Asset.class);
when(asset.getIdentifier()).thenReturn(assetId);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
index 8356406c31a..3841ed70afe 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestFlowResource.java
@@ -26,15 +26,22 @@ import com.fasterxml.jackson.databind.module.SimpleModule;
import io.prometheus.client.Collector.MetricFamilySamples.Sample;
import io.prometheus.client.CollectorRegistry;
import io.prometheus.client.exporter.common.TextFormat;
+import jakarta.ws.rs.HttpMethod;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.StreamingOutput;
import org.apache.nifi.authorization.AccessDeniedException;
+import org.apache.nifi.authorization.AuthorizableLookup;
import org.apache.nifi.authorization.AuthorizeAccess;
+import org.apache.nifi.authorization.Authorizer;
+import org.apache.nifi.authorization.RequestAction;
+import org.apache.nifi.authorization.resource.Authorizable;
+import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.validation.DisabledServiceValidationResult;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.controller.ProcessorNode;
+import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessingPerformanceStatus;
@@ -52,12 +59,15 @@ import org.apache.nifi.registry.flow.FlowVersionLocation;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.ResourceNotFoundException;
+import org.apache.nifi.web.Revision;
import org.apache.nifi.web.api.dto.ComponentDifferenceDTO;
import org.apache.nifi.web.api.dto.DifferenceDTO;
+import org.apache.nifi.web.api.dto.RevisionDTO;
import org.apache.nifi.web.api.entity.ActivateControllerServicesEntity;
import org.apache.nifi.web.api.entity.ClearBulletinsForGroupRequestEntity;
import org.apache.nifi.web.api.entity.ConnectorEntity;
import org.apache.nifi.web.api.entity.FlowComparisonEntity;
+import org.apache.nifi.web.api.entity.ScheduleComponentsEntity;
import org.apache.nifi.web.api.request.FlowMetricsProducer;
import org.apache.nifi.web.api.request.FlowMetricsReportingStrategy;
import org.jetbrains.annotations.NotNull;
@@ -95,13 +105,18 @@ import static
org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.nullable;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
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.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -145,6 +160,9 @@ public class TestFlowResource {
@Mock
private NiFiProperties properties;
+ @Mock
+ private Authorizer authorizer;
+
@Mock
private ConnectorResource connectorResource;
@@ -152,6 +170,7 @@ public class TestFlowResource {
public void setUp() {
lenient().when(properties.isNode()).thenReturn(Boolean.FALSE);
resource.properties = properties;
+ resource.setAuthorizer(authorizer);
resource.setConnectorResource(connectorResource);
}
@@ -616,6 +635,238 @@ public class TestFlowResource {
assertEquals(5, componentIds.size(), "Should have exactly 5 authorized
components");
}
+ @Test
+ public void testStopSourcesUsesFacadeToIdentifySources() {
+ final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
+ entity.setId(PROCESS_GROUP_ID);
+ entity.setState(ScheduledState.STOPPED.name());
+
+ when(properties.isNode()).thenReturn(false);
+ resource.httpServletRequest = new MockHttpServletRequest();
+
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ final Set<String> identifiedSources = Set.of("source-processor",
"remote-output", "public-input");
+
when(serviceFacade.findSourceComponentIds(processGroup)).thenReturn(identifiedSources);
+
+ final ArgumentCaptor<Function<ProcessGroup, Set<String>>>
revisionsCaptor = ArgumentCaptor.captor();
+ when(serviceFacade.getRevisionsFromGroup(eq(PROCESS_GROUP_ID),
revisionsCaptor.capture())).thenReturn(Set.of());
+ when(serviceFacade.scheduleComponents(eq(PROCESS_GROUP_ID),
eq(ScheduledState.STOPPED), any())).thenReturn(entity);
+
+ final Response response = resource.stopSources(PROCESS_GROUP_ID,
entity);
+
+ assertNotNull(response);
+ assertEquals(HttpURLConnection.HTTP_OK, response.getStatus());
+ assertTrue(entity.getComponents().isEmpty());
+ assertEquals(identifiedSources,
revisionsCaptor.getValue().apply(processGroup));
+ }
+
+ @Test
+ public void testStopSourcesUnauthorizedFlowDoesNotDiscoverOrSchedule() {
+ final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
+ entity.setId(PROCESS_GROUP_ID);
+ entity.setState(ScheduledState.STOPPED.name());
+
+ resource.httpServletRequest = new MockHttpServletRequest();
+
+ doThrow(new
AccessDeniedException("denied")).when(serviceFacade).authorizeAccess(any());
+
+ assertThrows(AccessDeniedException.class, () ->
resource.stopSources(PROCESS_GROUP_ID, entity));
+
+ verify(serviceFacade, never()).verifyStopSources(anyString());
+ verify(serviceFacade, never()).getRevisionsFromGroup(anyString(),
any());
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ verify(serviceFacade, never()).verifyScheduleComponents(anyString(),
any(), any());
+ }
+
+ @Test
+ public void testStopSourcesWithNoSourcesReturnsEmptyComponents() {
+ final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
+ entity.setId(PROCESS_GROUP_ID);
+ entity.setState(ScheduledState.STOPPED.name());
+
+ when(properties.isNode()).thenReturn(false);
+ resource.httpServletRequest = new MockHttpServletRequest();
+
+ when(serviceFacade.getRevisionsFromGroup(eq(PROCESS_GROUP_ID),
any())).thenReturn(Set.of());
+ when(serviceFacade.scheduleComponents(eq(PROCESS_GROUP_ID),
eq(ScheduledState.STOPPED), any())).thenReturn(entity);
+
+ final Response response = resource.stopSources(PROCESS_GROUP_ID,
entity);
+
+ assertEquals(HttpURLConnection.HTTP_OK, response.getStatus());
+ assertTrue(entity.getComponents().isEmpty());
+ verify(serviceFacade).scheduleComponents(eq(PROCESS_GROUP_ID),
eq(ScheduledState.STOPPED), eq(Map.of()));
+ }
+
+ @Test
+ public void
testStopSourcesStatelessProcessGroupDoesNotDiscoverOrSchedule() {
+ final RevisionDTO revision = new RevisionDTO();
+ revision.setVersion(1L);
+ final ScheduleComponentsEntity entity = new ScheduleComponentsEntity();
+ entity.setId(PROCESS_GROUP_ID);
+ entity.setState(ScheduledState.STOPPED.name());
+ entity.setComponents(Map.of("source-processor", revision));
+
+ doThrow(new IllegalStateException("Stateless Process
Group")).when(serviceFacade).verifyStopSources(PROCESS_GROUP_ID);
+
+ assertThrows(IllegalStateException.class, () ->
resource.stopSources(PROCESS_GROUP_ID, entity));
+
+ verify(serviceFacade, never()).getRevisionsFromGroup(anyString(),
any());
+ verify(serviceFacade).authorizeAccess(any());
+ verify(serviceFacade, never()).verifyScheduleComponents(anyString(),
any(), any());
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ }
+
+ @Test
+ public void testStopSourcesResponseIncludesStoppedComponentIds() {
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(PROCESS_GROUP_ID);
+ request.setState(ScheduledState.STOPPED.name());
+
+ when(properties.isNode()).thenReturn(false);
+ resource.httpServletRequest = new MockHttpServletRequest();
+
+ when(serviceFacade.getRevisionsFromGroup(eq(PROCESS_GROUP_ID), any()))
+ .thenReturn(Set.of(new Revision(1L, "client",
"source-processor")));
+
+ final ScheduleComponentsEntity facadeResponse = new
ScheduleComponentsEntity();
+ facadeResponse.setId(PROCESS_GROUP_ID);
+ facadeResponse.setState(ScheduledState.STOPPED.name());
+ when(serviceFacade.scheduleComponents(eq(PROCESS_GROUP_ID),
eq(ScheduledState.STOPPED), any())).thenReturn(facadeResponse);
+
+ final Response response = resource.stopSources(PROCESS_GROUP_ID,
request);
+
+ assertEquals(HttpURLConnection.HTTP_OK, response.getStatus());
+ final ScheduleComponentsEntity responseEntity =
(ScheduleComponentsEntity) response.getEntity();
+ assertNotNull(responseEntity.getComponents());
+ assertEquals(Set.of("source-processor"),
responseEntity.getComponents().keySet());
+ }
+
+ @Test
+ public void testStopSourcesIdentifiesSourcesBeforeReplicate() {
+ final FlowResource spyResource = spy(resource);
+ doReturn(true).when(spyResource).isReplicateRequest();
+
doReturn(Response.ok().build()).when(spyResource).replicate(anyString(),
any(ScheduleComponentsEntity.class));
+
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(PROCESS_GROUP_ID);
+ request.setState(ScheduledState.STOPPED.name());
+ when(serviceFacade.getRevisionsFromGroup(eq(PROCESS_GROUP_ID),
any())).thenReturn(Set.of(
+ new Revision(1L, "client", "first-source"),
+ new Revision(2L, "client", "second-source")
+ ));
+
+ final Response response = spyResource.stopSources(PROCESS_GROUP_ID,
request);
+
+ assertEquals(HttpURLConnection.HTTP_OK, response.getStatus());
+ assertEquals(Set.of("first-source", "second-source"),
request.getComponents().keySet());
+ verify(serviceFacade).verifyStopSources(PROCESS_GROUP_ID,
Set.of("first-source", "second-source"));
+ verify(serviceFacade, times(2)).authorizeAccess(any());
+ verify(spyResource).replicate(HttpMethod.PUT, request);
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ }
+
+ @Test
+ public void testStopSourcesRechecksIdentifiedSourcesBeforeScheduling() {
+ final RevisionDTO revisionDto = new RevisionDTO();
+ revisionDto.setClientId("client");
+ revisionDto.setVersion(1L);
+
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(PROCESS_GROUP_ID);
+ request.setState(ScheduledState.STOPPED.name());
+ request.setComponents(Map.of("source-processor", revisionDto));
+ resource.httpServletRequest = new MockHttpServletRequest();
+
+ doNothing().when(serviceFacade).verifyStopSources(PROCESS_GROUP_ID);
+ doNothing()
+ .doThrow(new IllegalStateException("Source components
changed"))
+ .when(serviceFacade).verifyStopSources(PROCESS_GROUP_ID,
Set.of("source-processor"));
+
+ assertThrows(IllegalStateException.class, () ->
resource.stopSources(PROCESS_GROUP_ID, request));
+
+ verify(serviceFacade, times(2)).verifyStopSources(PROCESS_GROUP_ID,
Set.of("source-processor"));
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ }
+
+ @Test
+ public void testStopSourcesAuthorizesSuppliedComponentsBeforeReplicate() {
+ final FlowResource spyResource = spy(resource);
+ doReturn(true).when(spyResource).isReplicateRequest();
+
doReturn(Response.ok().build()).when(spyResource).replicate(anyString(),
any(ScheduleComponentsEntity.class));
+
+ final RevisionDTO firstRevision = new RevisionDTO();
+ firstRevision.setClientId("client");
+ firstRevision.setVersion(1L);
+ final RevisionDTO secondRevision = new RevisionDTO();
+ secondRevision.setClientId("client");
+ secondRevision.setVersion(2L);
+
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(PROCESS_GROUP_ID);
+ request.setState(ScheduledState.STOPPED.name());
+ request.setComponents(Map.of("first-source", firstRevision,
"second-source", secondRevision));
+
+ final Response response = spyResource.stopSources(PROCESS_GROUP_ID,
request);
+
+ assertEquals(HttpURLConnection.HTTP_OK, response.getStatus());
+ final ArgumentCaptor<AuthorizeAccess> authorizeAccessCaptor =
ArgumentCaptor.captor();
+ verify(serviceFacade,
times(2)).authorizeAccess(authorizeAccessCaptor.capture());
+ final AuthorizableLookup lookup = mock(AuthorizableLookup.class);
+ final Authorizable flow = mock(Authorizable.class);
+ final Authorizable firstSource = mock(Authorizable.class);
+ final Authorizable secondSource = mock(Authorizable.class);
+ when(lookup.getFlow()).thenReturn(flow);
+
when(lookup.getLocalConnectable("first-source")).thenReturn(firstSource);
+
when(lookup.getLocalConnectable("second-source")).thenReturn(secondSource);
+
+ authorizeAccessCaptor.getAllValues().forEach(authorizeAccess ->
authorizeAccess.authorize(lookup));
+
+ verify(flow).authorize(eq(authorizer), eq(RequestAction.READ),
nullable(NiFiUser.class));
+ verify(firstSource).authorize(eq(authorizer), eq(RequestAction.WRITE),
nullable(NiFiUser.class));
+ verify(secondSource).authorize(eq(authorizer),
eq(RequestAction.WRITE), nullable(NiFiUser.class));
+ verify(serviceFacade).verifyStopSources(PROCESS_GROUP_ID,
Set.of("first-source", "second-source"));
+ verify(serviceFacade, never()).getRevisionsFromGroup(anyString(),
any());
+ verify(spyResource).replicate(eq(HttpMethod.PUT), eq(request));
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ assertEquals(Set.of("first-source", "second-source"),
request.getComponents().keySet());
+ }
+
+ @Test
+ public void
testStopSourcesUnauthorizedSuppliedComponentsDoesNotReplicate() {
+ final FlowResource spyResource = spy(resource);
+
+ final RevisionDTO revisionDto = new RevisionDTO();
+ revisionDto.setClientId("client");
+ revisionDto.setVersion(1L);
+
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(PROCESS_GROUP_ID);
+ request.setState(ScheduledState.STOPPED.name());
+ request.setComponents(Map.of("source-processor", revisionDto));
+
+ final AuthorizableLookup lookup = mock(AuthorizableLookup.class);
+ final Authorizable flow = mock(Authorizable.class);
+ final Authorizable sourceProcessor = mock(Authorizable.class);
+ when(lookup.getFlow()).thenReturn(flow);
+
when(lookup.getLocalConnectable("source-processor")).thenReturn(sourceProcessor);
+ doThrow(new AccessDeniedException("denied")).when(sourceProcessor)
+ .authorize(eq(authorizer), eq(RequestAction.WRITE),
nullable(NiFiUser.class));
+ doAnswer(invocation -> {
+ invocation.getArgument(0, AuthorizeAccess.class).authorize(lookup);
+ return null;
+ }).when(serviceFacade).authorizeAccess(any());
+
+ assertThrows(AccessDeniedException.class, () ->
spyResource.stopSources(PROCESS_GROUP_ID, request));
+
+ verify(serviceFacade, times(2)).authorizeAccess(any());
+ verify(flow).authorize(eq(authorizer), eq(RequestAction.READ),
nullable(NiFiUser.class));
+ verify(sourceProcessor).authorize(eq(authorizer),
eq(RequestAction.WRITE), nullable(NiFiUser.class));
+ verify(serviceFacade, never()).verifyStopSources(PROCESS_GROUP_ID,
Set.of("source-processor"));
+ verify(spyResource, never()).replicate(anyString(),
any(ScheduleComponentsEntity.class));
+ verify(serviceFacade, never()).scheduleComponents(anyString(), any(),
any());
+ verify(serviceFacade, never()).getRevisionsFromGroup(anyString(),
any());
+ }
+
@Test
public void testGetConnectors() {
final ConnectorEntity connectorEntity = new ConnectorEntity();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/DtoFactoryTest.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/DtoFactoryTest.java
index a6f4fb1cd81..5bd149c33f7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/DtoFactoryTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/DtoFactoryTest.java
@@ -26,6 +26,7 @@ import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
+import org.apache.nifi.connectable.Position;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.queue.LoadBalanceCompression;
@@ -33,7 +34,13 @@ import org.apache.nifi.controller.queue.LoadBalanceStrategy;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.controller.service.ControllerServiceState;
+import org.apache.nifi.controller.status.ProcessGroupStatus;
+import org.apache.nifi.flow.ExecutionEngine;
+import org.apache.nifi.groups.FlowFileConcurrency;
+import org.apache.nifi.groups.FlowFileOutboundPolicy;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.groups.ProcessGroupCounts;
+import org.apache.nifi.groups.StatelessGroupScheduledState;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarManifest;
@@ -51,6 +58,7 @@ import org.apache.nifi.registry.flow.FlowRegistryClientNode;
import org.apache.nifi.registry.flow.diff.DifferenceType;
import org.apache.nifi.registry.flow.diff.FlowDifference;
import org.apache.nifi.web.ResourceNotFoundException;
+import org.apache.nifi.web.api.dto.flow.ProcessGroupFlowDTO;
import org.apache.nifi.web.api.entity.AllowableValueEntity;
import org.apache.nifi.web.api.entity.ParameterContextReferenceEntity;
import org.apache.nifi.web.revision.RevisionManager;
@@ -1062,4 +1070,102 @@ public class DtoFactoryTest {
when(context.getParameterReferenceManager()).thenReturn(ParameterReferenceManager.EMPTY);
}
+ @Test
+ void
testCreateProcessGroupDtoSetsResolvedExecutionEngineForInheritedUnderStatelessParent()
{
+ final ProcessGroup group = stubProcessGroup(ExecutionEngine.INHERITED,
ExecutionEngine.STATELESS);
+
+ final ProcessGroupDTO dto =
newDtoFactoryForParameters().createProcessGroupDto(group, false);
+
+ assertEquals("INHERITED", dto.getExecutionEngine());
+ assertEquals("STATELESS", dto.getResolvedExecutionEngine());
+ }
+
+ @Test
+ void
testCreateProcessGroupDtoSetsResolvedExecutionEngineStandardForRootInherited() {
+ final ProcessGroup group = stubProcessGroup(ExecutionEngine.INHERITED,
ExecutionEngine.STANDARD);
+
+ final ProcessGroupDTO dto =
newDtoFactoryForParameters().createProcessGroupDto(group, false);
+
+ assertEquals("INHERITED", dto.getExecutionEngine());
+ assertEquals("STANDARD", dto.getResolvedExecutionEngine());
+ }
+
+ @Test
+ void
testCreateProcessGroupDtoSetsResolvedExecutionEngineForConfiguredStateless() {
+ final ProcessGroup group = stubProcessGroup(ExecutionEngine.STATELESS,
ExecutionEngine.STATELESS);
+
+ final ProcessGroupDTO dto =
newDtoFactoryForParameters().createProcessGroupDto(group, false);
+
+ assertEquals("STATELESS", dto.getExecutionEngine());
+ assertEquals("STATELESS", dto.getResolvedExecutionEngine());
+ }
+
+ @Test
+ void testCreateProcessGroupFlowDtoSetsResolvedExecutionEngine() {
+ final ProcessGroup group = stubProcessGroup(ExecutionEngine.INHERITED,
ExecutionEngine.STATELESS);
+ final ProcessGroupStatus groupStatus = mock(ProcessGroupStatus.class);
+
when(groupStatus.getProcessorStatus()).thenReturn(Collections.emptyList());
+
when(groupStatus.getConnectionStatus()).thenReturn(Collections.emptyList());
+
when(groupStatus.getProcessGroupStatus()).thenReturn(Collections.emptyList());
+
when(groupStatus.getRemoteProcessGroupStatus()).thenReturn(Collections.emptyList());
+
when(groupStatus.getInputPortStatus()).thenReturn(Collections.emptyList());
+
when(groupStatus.getOutputPortStatus()).thenReturn(Collections.emptyList());
+
+ final ProcessGroupFlowDTO dto =
newDtoFactoryForParameters().createProcessGroupFlowDto(
+ group,
+ groupStatus,
+ mock(RevisionManager.class),
+ ignored -> Collections.emptyList(),
+ false
+ );
+
+ assertEquals("STATELESS", dto.getResolvedExecutionEngine());
+ }
+
+ @Test
+ void testCopyProcessGroupDtoCopiesResolvedExecutionEngine() {
+ final ProcessGroupDTO original = new ProcessGroupDTO();
+ original.setContents(new FlowSnippetDTO());
+ original.setExecutionEngine("INHERITED");
+ original.setResolvedExecutionEngine("STATELESS");
+
+ final ProcessGroupDTO copy =
newDtoFactoryForParameters().copy(original, false);
+
+ assertEquals("INHERITED", copy.getExecutionEngine());
+ assertEquals("STATELESS", copy.getResolvedExecutionEngine());
+ }
+
+ private static ProcessGroup stubProcessGroup(final ExecutionEngine
configured, final ExecutionEngine resolved) {
+ final ProcessGroup group = mock(ProcessGroup.class);
+ when(group.getIdentifier()).thenReturn("pg-1");
+ when(group.getPosition()).thenReturn(new Position(0, 0));
+ when(group.getComments()).thenReturn("");
+ when(group.getName()).thenReturn("group");
+ when(group.getVersionedComponentId()).thenReturn(Optional.empty());
+ when(group.getVersionControlInformation()).thenReturn(null);
+
when(group.getFlowFileConcurrency()).thenReturn(FlowFileConcurrency.UNBOUNDED);
+
when(group.getFlowFileOutboundPolicy()).thenReturn(FlowFileOutboundPolicy.STREAM_WHEN_AVAILABLE);
+ when(group.getDefaultFlowFileExpiration()).thenReturn("0 sec");
+ when(group.getDefaultBackPressureObjectThreshold()).thenReturn(10000L);
+ when(group.getDefaultBackPressureDataSizeThreshold()).thenReturn("1
GB");
+ when(group.getLogFileSuffix()).thenReturn(null);
+
when(group.getStatelessScheduledState()).thenReturn(StatelessGroupScheduledState.STOPPED);
+ when(group.getExecutionEngine()).thenReturn(configured);
+ when(group.resolveExecutionEngine()).thenReturn(resolved);
+ when(group.getMaxConcurrentTasks()).thenReturn(1);
+ when(group.getStatelessFlowTimeout()).thenReturn("1 min");
+ when(group.getParameterContext()).thenReturn(null);
+ when(group.getParent()).thenReturn(null);
+ when(group.getCounts()).thenReturn(new ProcessGroupCounts(0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0));
+ when(group.getProcessors()).thenReturn(Collections.emptySet());
+ when(group.getConnections()).thenReturn(Collections.emptySet());
+ when(group.getLabels()).thenReturn(Collections.emptySet());
+ when(group.getFunnels()).thenReturn(Collections.emptySet());
+ when(group.getProcessGroups()).thenReturn(Collections.emptySet());
+
when(group.getRemoteProcessGroups()).thenReturn(Collections.emptySet());
+ when(group.getInputPorts()).thenReturn(Collections.emptySet());
+ when(group.getOutputPorts()).thenReturn(Collections.emptySet());
+ return group;
+ }
+
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/EntityFactoryTest.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/EntityFactoryTest.java
new file mode 100644
index 00000000000..2edf44b2122
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/dto/EntityFactoryTest.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.web.api.dto;
+
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class EntityFactoryTest {
+
+ @Test
+ void
testCreateProcessGroupEntityPromotesResolvedExecutionEngineWhenUnauthorized() {
+ final ProcessGroupDTO dto = new ProcessGroupDTO();
+ dto.setId("pg-1");
+ dto.setResolvedExecutionEngine("STATELESS");
+
+ final PermissionsDTO permissions = new PermissionsDTO();
+ permissions.setCanRead(false);
+ permissions.setCanWrite(false);
+
+ final ProcessGroupEntity entity = new
EntityFactory().createProcessGroupEntity(dto, null, permissions, null, null);
+
+ assertEquals("STATELESS", entity.getResolvedExecutionEngine());
+ assertNull(entity.getComponent());
+ }
+
+ @Test
+ void testCreateProcessGroupEntityIncludesComponentWhenAuthorized() {
+ final ProcessGroupDTO dto = new ProcessGroupDTO();
+ dto.setId("pg-1");
+ dto.setResolvedExecutionEngine("STANDARD");
+
+ final PermissionsDTO permissions = new PermissionsDTO();
+ permissions.setCanRead(true);
+ permissions.setCanWrite(true);
+
+ final ProcessGroupEntity entity = new
EntityFactory().createProcessGroupEntity(dto, null, permissions, null, null);
+
+ assertEquals("STANDARD", entity.getResolvedExecutionEngine());
+ assertEquals(dto, entity.getComponent());
+ }
+}
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.spec.ts
new file mode 100644
index 00000000000..8d8f7966894
--- /dev/null
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.spec.ts
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { TestBed } from '@angular/core/testing';
+import { MockStore, provideMockStore } from '@ngrx/store/testing';
+import type { Mock } from 'vitest';
+
+import { CanvasContextMenu } from './canvas-context-menu.service';
+import { CanvasUtils } from './canvas-utils.service';
+import { Client } from '../../../service/client.service';
+import { CanvasView } from './canvas-view.service';
+import { CanvasActionsService } from './canvas-actions.service';
+import { DraggableBehavior } from './behavior/draggable-behavior.service';
+import * as FlowActions from '../state/flow/flow.actions';
+import type { ResolvedExecutionEngine } from '../state/flow';
+import type { ContextMenuItemDefinition } from
'../../../ui/common/context-menu/context-menu.component';
+
+interface SetupOptions {
+ currentProcessGroupId?: string;
+ isProcessGroup?: boolean;
+ resolvedExecutionEngine?: ResolvedExecutionEngine;
+}
+
+function menuItem(menuItems: ContextMenuItemDefinition[], text: string):
ContextMenuItemDefinition {
+ const item = menuItems.find((candidate) => candidate.text === text);
+ if (!item) {
+ throw new Error(`Expected menu item "${text}"`);
+ }
+ return item;
+}
+
+async function setup(options: SetupOptions = {}) {
+ const currentProcessGroupId = options.currentProcessGroupId ??
'current-pg';
+ const canvasUtils = {
+ getProcessGroupId: vi.fn().mockReturnValue(currentProcessGroupId),
+ isProcessGroup: vi.fn().mockReturnValue(options.isProcessGroup ??
false),
+ getResolvedExecutionEngine:
vi.fn().mockReturnValue(options.resolvedExecutionEngine ?? 'STANDARD')
+ };
+
+ await TestBed.configureTestingModule({
+ providers: [
+ CanvasContextMenu,
+ provideMockStore(),
+ { provide: CanvasUtils, useValue: canvasUtils },
+ { provide: Client, useValue: {} },
+ { provide: CanvasView, useValue: {} },
+ {
+ provide: CanvasActionsService,
+ useValue: {
+ getConditionFunction: () => () => false,
+ getActionFunction: () => () => undefined
+ }
+ },
+ { provide: DraggableBehavior, useValue: {} }
+ ]
+ }).compileComponents();
+
+ const service = TestBed.inject(CanvasContextMenu);
+ const store = TestBed.inject(MockStore);
+ const dispatchSpy = vi.spyOn(store, 'dispatch') as Mock;
+
+ return { service, canvasUtils, dispatchSpy };
+}
+
+describe('CanvasContextMenu', () => {
+ describe('Stop Sources', () => {
+ it('is immediately after Stop in the root menu', async () => {
+ const { service } = await setup();
+ const menuItems = service.getMenu('root')!.menuItems;
+ const texts = menuItems.map((item) => item.text);
+ const stopIndex = texts.indexOf('Stop');
+ const stopSourcesIndex = texts.indexOf('Stop Sources');
+
+ expect(stopIndex).toBeGreaterThan(-1);
+ expect(stopSourcesIndex).toBe(stopIndex + 1);
+ });
+
+ it('is visible on an empty canvas and dispatches stopSources for the
current group', async () => {
+ const { service, canvasUtils, dispatchSpy } = await setup({
currentProcessGroupId: 'current-pg' });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = { empty: () => true };
+
+ expect(stopSources.condition!(selection as never)).toBe(true);
+ stopSources.action!(selection as never);
+
+ expect(canvasUtils.getProcessGroupId).toHaveBeenCalled();
+ expect(dispatchSpy).toHaveBeenCalledWith(FlowActions.stopSources({
request: { id: 'current-pg' } }));
+ });
+
+ it('is visible for a selected process group and dispatches stopSources
for that group', async () => {
+ const { service, dispatchSpy } = await setup({ isProcessGroup:
true });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = {
+ empty: () => false,
+ datum: () => ({ id: 'pg-child', resolvedExecutionEngine:
'STANDARD' })
+ };
+
+ expect(stopSources.condition!(selection as never)).toBe(true);
+ stopSources.action!(selection as never);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(FlowActions.stopSources({
request: { id: 'pg-child' } }));
+ });
+
+ it('is hidden when the selection is not a process group', async () => {
+ const { service } = await setup({ isProcessGroup: false });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = { empty: () => false };
+
+ expect(stopSources.condition!(selection as never)).toBe(false);
+ });
+
+ it('is hidden on an empty canvas when the current group resolves to
STATELESS', async () => {
+ const { service } = await setup({ resolvedExecutionEngine:
'STATELESS' });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = { empty: () => true };
+
+ expect(stopSources.condition!(selection as never)).toBe(false);
+ });
+
+ it('is hidden for a selected process group when the resolved engine is
missing', async () => {
+ const { service } = await setup({ isProcessGroup: true });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = {
+ empty: () => false,
+ datum: () => ({ id: 'pg-child' })
+ };
+
+ expect(stopSources.condition!(selection as never)).toBe(false);
+ });
+
+ it('is visible on an empty canvas when the current group is configured
INHERITED but resolves to STANDARD', async () => {
+ const { service } = await setup({ resolvedExecutionEngine:
'STANDARD' });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = { empty: () => true };
+
+ expect(stopSources.condition!(selection as never)).toBe(true);
+ });
+
+ it('is visible for a selected process group with no component when the
entity resolves to STANDARD', async () => {
+ const { service } = await setup({ isProcessGroup: true });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = {
+ empty: () => false,
+ datum: () => ({ id: 'pg-child', resolvedExecutionEngine:
'STANDARD' })
+ };
+
+ expect(stopSources.condition!(selection as never)).toBe(true);
+ });
+
+ it('is hidden for a selected process group with no component when the
entity resolves to STATELESS', async () => {
+ const { service } = await setup({ isProcessGroup: true });
+ const stopSources = menuItem(service.getMenu('root')!.menuItems,
'Stop Sources');
+ const selection = {
+ empty: () => false,
+ datum: () => ({ id: 'pg-child', resolvedExecutionEngine:
'STATELESS' })
+ };
+
+ expect(stopSources.condition!(selection as never)).toBe(false);
+ });
+ });
+});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts
index 5eb22074894..6267e71468f 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-context-menu.service.ts
@@ -47,6 +47,7 @@ import {
replayLastProvenanceEvent,
requestRefreshRemoteProcessGroup,
runOnce,
+ stopSources,
stopVersionControlRequest,
terminateThreads,
updatePositions
@@ -773,6 +774,36 @@ export class CanvasContextMenu implements
ContextMenuDefinitionProvider {
text: 'Stop',
action: this.canvasActionsService.getActionFunction('stop')
},
+ {
+ condition: (selection: any) => {
+ if (!(selection.empty() ||
this.canvasUtils.isProcessGroup(selection))) {
+ return false;
+ }
+ const resolved = selection.empty()
+ ? this.canvasUtils.getResolvedExecutionEngine()
+ : selection.datum().resolvedExecutionEngine;
+ return resolved === 'STANDARD';
+ },
+ clazz: 'fa fa-stop-circle-o',
+ text: 'Stop Sources',
+ action: (selection: any) => {
+ let processGroupId: string;
+ if (selection.empty()) {
+ processGroupId = this.canvasUtils.getProcessGroupId();
+ } else {
+ const selectionData = selection.datum();
+ processGroupId = selectionData.id;
+ }
+
+ this.store.dispatch(
+ stopSources({
+ request: {
+ id: processGroupId
+ }
+ })
+ );
+ }
+ },
{
condition: (selection: any) => {
if (selection.size() !== 1) {
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-utils.service.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-utils.service.ts
index b8847fc7271..8da73babd8e 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-utils.service.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/canvas-utils.service.ts
@@ -20,13 +20,15 @@ import * as d3 from 'd3';
import { humanizer, Humanizer } from 'humanize-duration';
import { Store } from '@ngrx/store';
import { CanvasState } from '../state';
+import type { ResolvedExecutionEngine } from '../state/flow';
import {
selectBreadcrumbs,
selectCanvasPermissions,
selectConnections,
selectCurrentParameterContext,
selectCurrentProcessGroupId,
- selectParentProcessGroupId
+ selectParentProcessGroupId,
+ selectResolvedExecutionEngine
} from '../state/flow/flow.selectors';
import { initialState as initialFlowState } from '../state/flow/flow.reducer';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';
@@ -80,6 +82,8 @@ export class CanvasUtils {
private trimLengthCaches: Map<string, Map<string, Map<number, number>>> =
new Map();
private currentProcessGroupId: string = initialFlowState.id;
private parentProcessGroupId: string | null =
initialFlowState.flow.processGroupFlow.parentGroupId;
+ private currentResolvedExecutionEngine: ResolvedExecutionEngine =
+ initialFlowState.flow.processGroupFlow.resolvedExecutionEngine;
private canvasPermissions: Permissions = initialFlowState.flow.permissions;
private currentUser: CurrentUser = initialUserState.user;
private currentParameterContext: ParameterContextReferenceEntity | null =
@@ -110,6 +114,13 @@ export class CanvasUtils {
this.parentProcessGroupId = parentProcessGroupId;
});
+ this.store
+ .select(selectResolvedExecutionEngine)
+ .pipe(takeUntilDestroyed(this.destroyRef))
+ .subscribe((resolvedExecutionEngine) => {
+ this.currentResolvedExecutionEngine = resolvedExecutionEngine;
+ });
+
this.store
.select(selectCanvasPermissions)
.pipe(takeUntilDestroyed(this.destroyRef))
@@ -229,6 +240,14 @@ export class CanvasUtils {
return this.currentProcessGroupId;
}
+ /**
+ * The Execution Engine that will actually run the current Process Group
+ * after resolving INHERITED. Always STANDARD or STATELESS.
+ */
+ public getResolvedExecutionEngine(): ResolvedExecutionEngine {
+ return this.currentResolvedExecutionEngine;
+ }
+
/**
* Returns the parent group id or null if current is root.
*/
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/flow.service.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/flow.service.ts
index b605afeda3f..1f5f469ab30 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/flow.service.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/service/flow.service.ts
@@ -43,6 +43,8 @@ import {
SaveToVersionControlRequest,
StartProcessGroupRequest,
StopProcessGroupRequest,
+ StopSourcesRequest,
+ StopSourcesResponse,
StopVersionControlRequest,
TerminateThreadsRequest,
UploadProcessGroupRequest,
@@ -403,6 +405,18 @@ export class FlowService implements
PropertyDescriptorRetriever {
return
this.httpClient.put(`${FlowService.API}/flow/process-groups/${request.id}`,
stopRequest);
}
+ stopSources(request: StopSourcesRequest): Observable<StopSourcesResponse> {
+ const stopRequest: ProcessGroupRunStatusRequest = {
+ id: request.id,
+ disconnectedNodeAcknowledged:
this.clusterConnectionService.isDisconnectionAcknowledged(),
+ state: 'STOPPED'
+ };
+ return this.httpClient.put<StopSourcesResponse>(
+ `${FlowService.API}/flow/process-groups/${request.id}/sources`,
+ stopRequest
+ );
+ }
+
stopRemoteProcessGroupsInProcessGroup(request: StopProcessGroupRequest):
Observable<any> {
const stopRequest = {
disconnectedNodeAcknowledged:
this.clusterConnectionService.isDisconnectionAcknowledged(),
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts
index 76587c6a0cc..441f5b5c5e8 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.actions.ts
@@ -92,6 +92,7 @@ import {
StopComponentsRequest,
StopProcessGroupRequest,
StopProcessGroupResponse,
+ StopSourcesRequest,
StopVersionControlRequest,
StopVersionControlResponse,
TerminateThreadsRequest,
@@ -749,6 +750,13 @@ export const startCurrentProcessGroup =
createAction(`${CANVAS_PREFIX} Start Cur
export const stopCurrentProcessGroup = createAction(`${CANVAS_PREFIX} Stop
Current Process Group`);
+export const stopSources = createAction(`${CANVAS_PREFIX} Stop Sources`,
props<{ request: StopSourcesRequest }>());
+
+export const stopSourcesSuccess = createAction(
+ `${CANVAS_PREFIX} Stop Sources Success`,
+ props<{ response: StopProcessGroupResponse }>()
+);
+
export const enableControllerServicesInCurrentProcessGroup = createAction(
`${CANVAS_PREFIX} Enable Controller Services In Current Process Group`
);
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.spec.ts
index f8ffd720ac8..4dc36a34dd7 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.spec.ts
@@ -17,7 +17,7 @@
import { FlowService } from '../../service/flow.service';
import * as FlowActions from './flow.actions';
-import { of, ReplaySubject, take, throwError } from 'rxjs';
+import { firstValueFrom, of, ReplaySubject, Subject, take, throwError, toArray
} from 'rxjs';
import { MatDialog, MatDialogRef } from '@angular/material/dialog';
import { ComponentHistoryEntity } from '../../../../state/shared';
import { EditProcessor } from
'../../../../ui/common/component-dialogs/edit-processor/edit-processor.component';
@@ -35,7 +35,8 @@ import {
CreateComponentResponse,
CreateConnection,
flowFeatureKey,
- MoveToFrontRequest
+ MoveToFrontRequest,
+ StopSourcesResponse
} from './index';
import {
BacklogRequestEntity,
@@ -81,7 +82,6 @@ import * as fromParameter from
'../parameter/parameter.reducer';
import { flowAnalysisFeatureKey } from '../flow-analysis';
import * as fromFlowAnalysis from '../flow-analysis/flow-analysis.reducer';
import * as EmptyQueueActions from
'../../../../state/empty-queue/empty-queue.actions';
-import { firstValueFrom } from 'rxjs';
describe('FlowEffects', () => {
let action$: ReplaySubject<Action>;
@@ -834,7 +834,8 @@ describe('FlowEffects', () => {
createConnection: vi.fn(),
createLabel: vi.fn(),
clearBulletinsForProcessGroup: vi.fn(),
- submitProcessorBacklogRequest: vi.fn()
+ submitProcessorBacklogRequest: vi.fn(),
+ stopSources: vi.fn()
}
},
{
@@ -1127,6 +1128,139 @@ describe('FlowEffects', () => {
});
});
+ describe('stopSources$', () => {
+ beforeEach(() => {
+ effects = TestBed.inject(FlowEffects);
+ });
+
+ it('should call flowService.stopSources and dispatch
stopSourcesSuccess', async () => {
+ vi.spyOn(flowService, 'stopSources').mockReturnValue(
+ of({ id: 'test-group-id', state: 'STOPPED', components: {} })
+ );
+
+ action$.next(FlowActions.stopSources({ request: { id:
'test-group-id' } }));
+
+ const result = await new Promise((resolve) =>
effects.stopSources$.pipe(take(1)).subscribe(resolve));
+
+ expect(result).toEqual(
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: { id: 'test-group-id', state: 'STOPPED' }
+ }
+ })
+ );
+ expect(flowService.stopSources).toHaveBeenCalledWith({ id:
'test-group-id' });
+ });
+
+ it('should keep overlapping stop sources requests active', async () =>
{
+ const firstResponse$ = new Subject<StopSourcesResponse>();
+ const secondResponse$ = new Subject<StopSourcesResponse>();
+ vi.spyOn(flowService, 'stopSources').mockImplementation((request)
=>
+ request.id === 'first-group-id' ? firstResponse$ :
secondResponse$
+ );
+ const resultsPromise =
firstValueFrom(effects.stopSources$.pipe(take(2), toArray()));
+
+ action$.next(FlowActions.stopSources({ request: { id:
'first-group-id' } }));
+ action$.next(FlowActions.stopSources({ request: { id:
'second-group-id' } }));
+ secondResponse$.next({ id: 'second-group-id', state: 'STOPPED',
components: {} });
+ secondResponse$.complete();
+ firstResponse$.next({ id: 'first-group-id', state: 'STOPPED',
components: {} });
+ firstResponse$.complete();
+
+ expect(await resultsPromise).toEqual([
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: { id: 'second-group-id', state: 'STOPPED' }
+ }
+ }),
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: { id: 'first-group-id', state: 'STOPPED' }
+ }
+ })
+ ]);
+ expect(flowService.stopSources).toHaveBeenCalledTimes(2);
+ });
+
+ it('should dispatch flowSnackbarError and not stopSourcesSuccess when
stopSources fails', async () => {
+ const errorHelper = TestBed.inject(ErrorHelper);
+ const errorResponse = new HttpErrorResponse({
+ error: 'stop sources failed',
+ status: 409,
+ statusText: 'Conflict'
+ });
+ vi.spyOn(flowService, 'stopSources').mockReturnValue(throwError(()
=> errorResponse));
+ vi.spyOn(errorHelper, 'getErrorString').mockReturnValue('Formatted
error message');
+
+ action$.next(FlowActions.stopSources({ request: { id:
'test-group-id' } }));
+
+ const result = await new Promise((resolve) =>
effects.stopSources$.pipe(take(1)).subscribe(resolve));
+
+ expect(result).toEqual(FlowActions.flowSnackbarError({ error:
'Formatted error message' }));
+ expect((result as
Action).type).not.toBe(FlowActions.stopSourcesSuccess.type);
+ expect(flowService.stopSources).toHaveBeenCalledWith({ id:
'test-group-id' });
+ });
+ });
+
+ describe('stopSourcesCurrentProcessGroupSuccess$', () => {
+ beforeEach(() => {
+ effects = TestBed.inject(FlowEffects);
+ });
+
+ it('should dispatch reloadFlow when sources are stopped in the current
process group', async () => {
+ store.overrideSelector(selectCurrentProcessGroupId,
'test-group-id');
+ store.refreshState();
+
+ action$.next(
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: { id: 'test-group-id', state: 'STOPPED' }
+ }
+ })
+ );
+
+ const result = await new Promise((resolve) =>
+
effects.stopSourcesCurrentProcessGroupSuccess$.pipe(take(1)).subscribe(resolve)
+ );
+
+ expect(result).toEqual(FlowActions.reloadFlow());
+ });
+ });
+
+ describe('stopSourcesSuccess$', () => {
+ beforeEach(() => {
+ effects = TestBed.inject(FlowEffects);
+ });
+
+ it('should dispatch loadChildProcessGroup when sources are stopped in
a child process group', async () => {
+ store.overrideSelector(selectCurrentProcessGroupId, 'current-pg');
+ store.refreshState();
+
+ action$.next(
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: { id: 'pg-child', state: 'STOPPED' }
+ }
+ })
+ );
+
+ const result = await new Promise((resolve) =>
effects.stopSourcesSuccess$.pipe(take(1)).subscribe(resolve));
+
+ expect(result).toEqual(
+ FlowActions.loadChildProcessGroup({
+ request: {
+ id: 'pg-child'
+ }
+ })
+ );
+ });
+ });
+
describe('clearBulletinsForProcessGroup$', () => {
beforeEach(() => {
effects = TestBed.inject(FlowEffects);
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts
index 5413002cecf..5d9220f4699 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.effects.ts
@@ -3656,6 +3656,64 @@ export class FlowEffects {
)
);
+ stopSources$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(FlowActions.stopSources),
+ map((action) => action.request),
+ mergeMap((request) =>
+ from(this.flowService.stopSources(request)).pipe(
+ map((response) =>
+ FlowActions.stopSourcesSuccess({
+ response: {
+ type: ComponentType.ProcessGroup,
+ component: {
+ id: response.id,
+ state: response.state
+ }
+ }
+ })
+ ),
+ catchError((errorResponse: HttpErrorResponse) =>
of(this.snackBarOrFullScreenError(errorResponse)))
+ )
+ )
+ )
+ );
+
+ /**
+ * If sources were stopped in the current process group, reload the flow
+ */
+ stopSourcesCurrentProcessGroupSuccess$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(FlowActions.stopSourcesSuccess),
+ map((action) => action.response),
+ concatLatestFrom(() =>
this.store.select(selectCurrentProcessGroupId)),
+ filter(([response, currentPg]) => response.component.id ===
currentPg),
+ switchMap(() => of(FlowActions.reloadFlow()))
+ )
+ );
+
+ /**
+ * If sources were stopped in a child ProcessGroup, reload that row; the
+ * schedule response does not contain all the displayed info
+ */
+ stopSourcesSuccess$ = createEffect(() =>
+ this.actions$.pipe(
+ ofType(FlowActions.stopSourcesSuccess),
+ map((action) => action.response),
+ concatLatestFrom(() =>
this.store.select(selectCurrentProcessGroupId)),
+ filter(([response, currentPg]) => response.component.id !==
currentPg),
+ switchMap(([response]) =>
+ of(
+ FlowActions.loadChildProcessGroup({
+ request: {
+ id: response.component.id
+ }
+ })
+ )
+ )
+ )
+ );
+
enableControllerServicesInCurrentProcessGroup$ = createEffect(() =>
this.actions$.pipe(
ofType(FlowActions.enableControllerServicesInCurrentProcessGroup),
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.reducer.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.reducer.ts
index f9d0122704c..6e6f602ef42 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.reducer.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.reducer.ts
@@ -73,6 +73,8 @@ import {
startComponentSuccess,
startPollingProcessorUntilStopped,
startProcessGroupSuccess,
+ stopSources,
+ stopSourcesSuccess,
startRemoteProcessGroupPolling,
stopComponent,
stopComponentSuccess,
@@ -123,6 +125,7 @@ export const initialState: FlowState = {
}
},
parameterContext: null,
+ resolvedExecutionEngine: 'STANDARD',
flow: {
processGroups: [],
remoteProcessGroups: [],
@@ -432,6 +435,7 @@ export const flowReducer = createReducer(
disableComponent,
startComponent,
stopComponent,
+ stopSources,
runOnce,
(state) => ({
...state,
@@ -443,6 +447,7 @@ export const flowReducer = createReducer(
disableProcessGroupSuccess,
startProcessGroupSuccess,
stopProcessGroupSuccess,
+ stopSourcesSuccess,
(state) => ({
...state,
saving: false
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.selectors.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.selectors.ts
index 27b84bddbb0..a699a24460a 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.selectors.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/flow.selectors.ts
@@ -70,6 +70,11 @@ export const selectParentProcessGroupId = createSelector(
(state: FlowState) => state.flow.processGroupFlow.parentGroupId
);
+export const selectResolvedExecutionEngine = createSelector(
+ selectFlowState,
+ (state: FlowState) => state.flow.processGroupFlow.resolvedExecutionEngine
+);
+
export const selectProcessGroupIdFromRoute =
createSelector(selectCurrentRoute, (route) => {
if (route) {
// always select the process group from the route
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/index.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/index.ts
index c4e821d0152..1b7b830ec22 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/index.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/flow-designer/state/flow/index.ts
@@ -470,6 +470,8 @@ export interface Flow {
funnels: ComponentEntity[];
}
+export type ResolvedExecutionEngine = 'STANDARD' | 'STATELESS';
+
export interface ProcessGroupFlow {
id: string;
uri: string;
@@ -478,6 +480,7 @@ export interface ProcessGroupFlow {
parameterContext: ParameterContextReferenceEntity | null;
flow: Flow;
lastRefreshed: string;
+ resolvedExecutionEngine: ResolvedExecutionEngine;
}
export interface ProcessGroupFlowEntity {
@@ -648,6 +651,17 @@ export interface StopProcessGroupRequest {
errorStrategy: 'snackbar' | 'banner';
}
+export interface StopSourcesRequest {
+ id: string;
+}
+
+export interface StopSourcesResponse {
+ id: string;
+ state: 'STOPPED';
+ components: Record<string, Revision>;
+ disconnectedNodeAcknowledged?: boolean;
+}
+
export interface StopComponentResponse {
type: ComponentType;
component: ComponentEntity;
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusteredStopSourcesIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusteredStopSourcesIT.java
new file mode 100644
index 00000000000..3b5b2c9e1ec
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusteredStopSourcesIT.java
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.tests.system.pg;
+
+import jakarta.ws.rs.WebApplicationException;
+import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.groups.StatelessGroupScheduledState;
+import org.apache.nifi.tests.system.NiFiInstanceFactory;
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.apache.nifi.web.api.entity.ScheduleComponentsEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.fail;
+
+public class ClusteredStopSourcesIT extends NiFiSystemIT {
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ return createTwoNodeInstanceFactory();
+ }
+
+ @Test
+ public void testStopSourcesSkipsStatelessDescendants() throws
NiFiClientException, IOException, InterruptedException {
+ final MixedEngineFlow flow = createMixedEngineFlow();
+ startMixedEngineFlow(flow);
+
+ final ScheduleComponentsEntity response =
getNifiClient().getFlowClient()
+ .stopProcessGroupSources(flow.parentGroup().getId(),
stopSourcesRequest(flow.parentGroup().getId()));
+
+ assertEquals(Set.of(flow.standardSource().getId()),
response.getComponents().keySet());
+ assertProcessorStateOnAllNodes(flow.standardSource().getId(),
ScheduledState.STOPPED);
+ assertProcessorStateOnAllNodes(flow.standardDownstream().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessSource().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessDownstream().getId(),
ScheduledState.RUNNING);
+ assertStatelessGroupStateOnAllNodes(flow.statelessGroup().getId(),
StatelessGroupScheduledState.RUNNING);
+ }
+
+ @Test
+ public void testStopSourcesRejectsStatelessProcessGroup() throws
NiFiClientException, IOException, InterruptedException {
+ final MixedEngineFlow flow = createMixedEngineFlow();
+ startMixedEngineFlow(flow);
+
+ final NiFiClientException exception =
assertThrows(NiFiClientException.class, () -> getNifiClient().getFlowClient()
+ .stopProcessGroupSources(flow.statelessGroup().getId(),
stopSourcesRequest(flow.statelessGroup().getId())));
+
+ assertConflict(exception);
+ assertProcessorStateOnAllNodes(flow.statelessSource().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessDownstream().getId(),
ScheduledState.RUNNING);
+ assertStatelessGroupStateOnAllNodes(flow.statelessGroup().getId(),
StatelessGroupScheduledState.RUNNING);
+ }
+
+ @Test
+ public void testStopSourcesRejectsMismatchedComponentIds() throws
NiFiClientException, IOException, InterruptedException {
+ final MixedEngineFlow flow = createMixedEngineFlow();
+ startMixedEngineFlow(flow);
+ final ScheduleComponentsEntity request =
stopSourcesRequest(flow.parentGroup().getId());
+ request.setComponents(Map.of());
+
+ final NiFiClientException exception =
assertThrows(NiFiClientException.class, () -> getNifiClient().getFlowClient()
+ .stopProcessGroupSources(flow.parentGroup().getId(), request));
+
+ assertConflict(exception);
+ assertProcessorStateOnAllNodes(flow.standardSource().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.standardDownstream().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessSource().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessDownstream().getId(),
ScheduledState.RUNNING);
+ assertStatelessGroupStateOnAllNodes(flow.statelessGroup().getId(),
StatelessGroupScheduledState.RUNNING);
+ }
+
+ @Test
+ public void testStopSourcesRejectsWhenNodeIdentifiesDifferentSources()
throws NiFiClientException, IOException, InterruptedException {
+ final MixedEngineFlow flow = createMixedEngineFlow();
+ startMixedEngineFlow(flow);
+ final long node1Revision =
stopAndRestartProcessorOnNode(flow.standardSource().getId(), 1);
+ final long node2Revision =
stopAndRenameProcessorOnNode(flow.standardSource().getId(), 2);
+
+ assertEquals(node1Revision, node2Revision, "Processor revisions must
match so source-set verification rejects the request");
+ assertProcessorStateOnNode(flow.standardSource().getId(),
ScheduledState.RUNNING, 1);
+ assertProcessorStateOnNode(flow.standardSource().getId(),
ScheduledState.STOPPED, 2);
+
+ final NiFiClientException exception =
assertThrows(NiFiClientException.class, () -> getNifiClient().getFlowClient()
+ .stopProcessGroupSources(flow.parentGroup().getId(),
stopSourcesRequest(flow.parentGroup().getId())));
+
+ assertConflict(exception);
+ assertProcessorStateOnNode(flow.standardSource().getId(),
ScheduledState.RUNNING, 1);
+ assertProcessorStateOnNode(flow.standardSource().getId(),
ScheduledState.STOPPED, 2);
+ assertProcessorStateOnAllNodes(flow.standardDownstream().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessSource().getId(),
ScheduledState.RUNNING);
+ assertProcessorStateOnAllNodes(flow.statelessDownstream().getId(),
ScheduledState.RUNNING);
+ assertStatelessGroupStateOnAllNodes(flow.statelessGroup().getId(),
StatelessGroupScheduledState.RUNNING);
+ }
+
+ private MixedEngineFlow createMixedEngineFlow() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity parentGroup =
getClientUtil().createProcessGroup("Parent", "root");
+ final ProcessGroupEntity standardGroup =
getClientUtil().createProcessGroup("Standard", parentGroup.getId());
+ final ProcessGroupEntity statelessGroup =
getClientUtil().createProcessGroup("Stateless", parentGroup.getId());
+ getClientUtil().markStateless(statelessGroup, "1 min");
+
+ final ProcessorEntity standardSource =
getClientUtil().createProcessor(GENERATE_FLOWFILE, standardGroup.getId());
+ final ProcessorEntity standardDownstream =
getClientUtil().createProcessor(TERMINATE_FLOWFILE, standardGroup.getId());
+ getClientUtil().createConnection(standardSource, standardDownstream,
SUCCESS, standardGroup.getId());
+
+ final ProcessorEntity statelessSource =
getClientUtil().createProcessor(GENERATE_FLOWFILE, statelessGroup.getId());
+ final ProcessorEntity statelessDownstream =
getClientUtil().createProcessor(TERMINATE_FLOWFILE, statelessGroup.getId());
+ getClientUtil().createConnection(statelessSource, statelessDownstream,
SUCCESS, statelessGroup.getId());
+
+ getClientUtil().waitForValidProcessor(standardSource.getId());
+ getClientUtil().waitForValidProcessor(standardDownstream.getId());
+ getClientUtil().waitForValidProcessor(statelessSource.getId());
+ getClientUtil().waitForValidProcessor(statelessDownstream.getId());
+
+ return new MixedEngineFlow(parentGroup, standardGroup, statelessGroup,
+ standardSource, standardDownstream, statelessSource,
statelessDownstream);
+ }
+
+ private void startMixedEngineFlow(final MixedEngineFlow flow) throws
NiFiClientException, IOException, InterruptedException {
+
getClientUtil().startProcessGroupComponents(flow.standardGroup().getId());
+
getClientUtil().startProcessGroupComponents(flow.statelessGroup().getId());
+ getClientUtil().waitForRunningProcessor(flow.standardSource().getId());
+
getClientUtil().waitForRunningProcessor(flow.standardDownstream().getId());
+
getClientUtil().waitForRunningProcessor(flow.statelessSource().getId());
+
getClientUtil().waitForRunningProcessor(flow.statelessDownstream().getId());
+ }
+
+ private ScheduleComponentsEntity stopSourcesRequest(final String groupId) {
+ final ScheduleComponentsEntity request = new
ScheduleComponentsEntity();
+ request.setId(groupId);
+ request.setState(ScheduledState.STOPPED.name());
+ request.setDisconnectedNodeAcknowledged(true);
+ return request;
+ }
+
+ private long stopAndRestartProcessorOnNode(final String processorId, final
int nodeIndex)
+ throws NiFiClientException, IOException, InterruptedException {
+ try {
+ switchClientToNode(nodeIndex);
+ final ProcessorEntity processor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ processor.setDisconnectedNodeAcknowledged(true);
+
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).stopProcessor(processor);
+ waitForProcessorStateOnCurrentNode(processorId,
ScheduledState.STOPPED);
+
+ final ProcessorEntity stoppedProcessor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ stoppedProcessor.setDisconnectedNodeAcknowledged(true);
+ final ProcessorEntity restartedProcessor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).startProcessor(stoppedProcessor);
+ return restartedProcessor.getRevision().getVersion();
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private long stopAndRenameProcessorOnNode(final String processorId, final
int nodeIndex)
+ throws NiFiClientException, IOException, InterruptedException {
+ try {
+ switchClientToNode(nodeIndex);
+ final ProcessorEntity processor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ processor.setDisconnectedNodeAcknowledged(true);
+
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).stopProcessor(processor);
+ waitForProcessorStateOnCurrentNode(processorId,
ScheduledState.STOPPED);
+
+ final ProcessorEntity stoppedProcessor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+
stoppedProcessor.getComponent().setName(stoppedProcessor.getComponent().getName()
+ " Node " + nodeIndex);
+ stoppedProcessor.setDisconnectedNodeAcknowledged(true);
+ final ProcessorEntity renamedProcessor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).updateProcessor(stoppedProcessor);
+ return renamedProcessor.getRevision().getVersion();
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private void waitForProcessorStateOnCurrentNode(final String processorId,
final ScheduledState expectedState) throws InterruptedException {
+ waitFor(() -> {
+ final ProcessorEntity processor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ return
expectedState.name().equals(processor.getComponent().getState())
+ &&
expectedState.name().equals(processor.getComponent().getPhysicalState());
+ });
+ }
+
+ private void assertProcessorStateOnNode(final String processorId, final
ScheduledState expectedState, final int nodeIndex)
+ throws NiFiClientException, IOException {
+ try {
+ switchClientToNode(nodeIndex);
+ final ProcessorEntity processor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ assertEquals(expectedState.name(),
processor.getComponent().getState(),
+ "Unexpected state for Processor %s on Node
%d".formatted(processorId, nodeIndex));
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private void assertProcessorStateOnAllNodes(final String processorId,
final ScheduledState expectedState)
+ throws NiFiClientException, IOException {
+ try {
+ for (int nodeIndex = 1; nodeIndex <= 2; nodeIndex++) {
+ switchClientToNode(nodeIndex);
+ final ProcessorEntity processor =
getNifiClient().getProcessorClient(DO_NOT_REPLICATE).getProcessor(processorId);
+ assertEquals(expectedState.name(),
processor.getComponent().getState(),
+ "Unexpected state for Processor %s on Node
%d".formatted(processorId, nodeIndex));
+ }
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private void assertStatelessGroupStateOnAllNodes(final String groupId,
final StatelessGroupScheduledState expectedState)
+ throws NiFiClientException, IOException {
+ try {
+ for (int nodeIndex = 1; nodeIndex <= 2; nodeIndex++) {
+ switchClientToNode(nodeIndex);
+ final ProcessGroupEntity group =
getNifiClient().getProcessGroupClient(DO_NOT_REPLICATE).getProcessGroup(groupId);
+ assertEquals(expectedState.name(),
group.getComponent().getStatelessGroupScheduledState(),
+ "Unexpected state for Stateless Process Group %s on
Node %d".formatted(groupId, nodeIndex));
+ }
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private void assertConflict(final NiFiClientException exception) {
+ final Throwable cause = exception.getCause();
+ if (cause instanceof final WebApplicationException
webApplicationException) {
+ assertEquals(409,
webApplicationException.getResponse().getStatus());
+ return;
+ }
+
+ fail("Expected WebApplicationException 409, got: " + cause);
+ }
+
+ private record MixedEngineFlow(
+ ProcessGroupEntity parentGroup,
+ ProcessGroupEntity standardGroup,
+ ProcessGroupEntity statelessGroup,
+ ProcessorEntity standardSource,
+ ProcessorEntity standardDownstream,
+ ProcessorEntity statelessSource,
+ ProcessorEntity statelessDownstream) {
+ }
+}
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/FlowClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/FlowClient.java
index 5218f17bf3c..a2c0818f4f3 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/FlowClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/FlowClient.java
@@ -85,6 +85,16 @@ public interface FlowClient {
ScheduleComponentsEntity scheduleProcessGroupComponents(
String processGroupId, ScheduleComponentsEntity
scheduleComponentsEntity) throws NiFiClientException, IOException;
+ /**
+ * Stops source components in a process group.
+ *
+ * @param processGroupId the id of a process group
+ * @param scheduleComponentsEntity the scheduled state to update to
+ * @return the entity representing the stopped source components
+ */
+ ScheduleComponentsEntity stopProcessGroupSources(
+ String processGroupId, ScheduleComponentsEntity
scheduleComponentsEntity) throws NiFiClientException, IOException;
+
/**
* Gets the possible versions for the given flow in the given bucket in the
* given registry.
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyFlowClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyFlowClient.java
index acf0a700a53..8a684cbfa9e 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyFlowClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyFlowClient.java
@@ -166,6 +166,32 @@ public class JerseyFlowClient extends AbstractJerseyClient
implements FlowClient
});
}
+ @Override
+ public ScheduleComponentsEntity stopProcessGroupSources(
+ final String processGroupId, final ScheduleComponentsEntity
scheduleComponentsEntity)
+ throws NiFiClientException, IOException {
+
+ if (StringUtils.isBlank(processGroupId)) {
+ throw new IllegalArgumentException("Process group id cannot be
null");
+ }
+
+ if (scheduleComponentsEntity == null) {
+ throw new IllegalArgumentException("ScheduleComponentsEntity
cannot be null");
+ }
+
+ scheduleComponentsEntity.setId(processGroupId);
+
+ return executeAction("Error stopping process group sources", () -> {
+ final WebTarget target = flowTarget
+ .path("process-groups/{id}/sources")
+ .resolveTemplate("id", processGroupId);
+
+ return getRequestBuilder(target).put(
+ Entity.entity(scheduleComponentsEntity,
MediaType.APPLICATION_JSON_TYPE),
+ ScheduleComponentsEntity.class);
+ });
+ }
+
@Override
public VersionedFlowSnapshotMetadataSetEntity getVersions(final String
registryId, final String bucketId, final String flowId)
throws NiFiClientException, IOException {
