NIFI-16337: Add support for stopping Process Group sources

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 void setExecutionEngine(final String executionEngine) {
         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() {

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 ParameterContextReferenceEntity getParameterContext() {
     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;
+    }
 }

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 String getProcessGroupUpdateStrategy() {
     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;
+    }
 }

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.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 StandardHttpResponseMapper(final NiFiProperties nifiProperties) {
         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());

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());
+            }
+        }
+    }
+}

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;
+    }
+}

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

@@ -387,6 +387,36 @@ Set<ControllerServiceEntity> getConnectorControllerServices(String connectorId,
      */
     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.
      *

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.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.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 Set<Revision> getRevisionsFromGroup(final String groupId, final Function<
         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);