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 11c520e61b7 NIFI-16487 Add Connector process group export support
(#11836)
11c520e61b7 is described below
commit 11c520e61b74c6ec2a5c8e9255cbca768cafc759
Author: Matt Gilman <[email protected]>
AuthorDate: Fri Oct 23 10:20:41 2026 -0400
NIFI-16487 Add Connector process group export support (#11836)
* NIFI-16487 Add Connector process group export support
Add Connector-scoped flow definition downloads, including component state,
cluster response merging, toolkit support, and Connector canvas actions.
* NIFI-16487: Addressing review feedback.
This closes #11836
---
.../ExportProcessGroupEndpointMerger.java | 5 +-
.../ExportProcessGroupEndpointMergerTest.java | 109 ++++++++++++++++
.../org/apache/nifi/web/NiFiServiceFacade.java | 13 ++
.../apache/nifi/web/StandardNiFiServiceFacade.java | 21 +++-
.../org/apache/nifi/web/api/ConnectorResource.java | 62 +++++++++
.../nifi/web/StandardNiFiServiceFacadeTest.java | 87 +++++++++++++
.../apache/nifi/web/api/TestConnectorResource.java | 50 ++++++++
.../connectors/service/connector.service.spec.ts | 80 ++++++++++++
.../pages/connectors/service/connector.service.ts | 11 +-
.../connector-canvas/connector-canvas.actions.ts | 11 ++
.../connector-canvas.effects.spec.ts | 21 ++++
.../connector-canvas/connector-canvas.effects.ts | 9 ++
.../connector-canvas.component.spec.ts | 139 ++++++++++++++++++++-
.../connector-canvas/connector-canvas.component.ts | 65 ++++++++++
.../connectors/ConnectorTroubleshootingIT.java | 31 +++++
.../nifi/toolkit/client/ConnectorClient.java | 29 +++++
.../toolkit/client/impl/JerseyConnectorClient.java | 44 +++++++
17 files changed, 778 insertions(+), 9 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMerger.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMerger.java
index 22fdd751fba..9a7eba7bfb0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMerger.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMerger.java
@@ -36,7 +36,7 @@ import java.util.Set;
import java.util.regex.Pattern;
/**
- * Merges responses for {@code GET /nifi-api/process-groups/{uuid}/download}.
+ * Merges responses for Process Group flow definition download endpoints.
* When component state is included in the export, each cluster node
contributes its own LOCAL state
* keyed by ordinal position. This merger combines the {@code localNodeStates}
maps from every node's
* snapshot into a single response.
@@ -44,7 +44,8 @@ import java.util.regex.Pattern;
public class ExportProcessGroupEndpointMerger implements
EndpointResponseMerger {
public static final Pattern PROCESS_GROUP_DOWNLOAD_URI_PATTERN =
-
Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/download");
+
Pattern.compile("/nifi-api/(?:process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))"
+ +
"|connectors/[a-f0-9\\-]{36}/process-groups/[a-f0-9\\-]{36})/download");
@Override
public boolean canHandle(final URI uri, final String method) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMergerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMergerTest.java
new file mode 100644
index 00000000000..26a12f87bc8
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMergerTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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 jakarta.ws.rs.core.Response;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedNodeState;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ExportProcessGroupEndpointMergerTest {
+
+ private static final String CONNECTOR_ID =
"12345678-1234-1234-1234-123456789012";
+ private static final String PROCESS_GROUP_ID =
"abcdef01-2345-6789-abcd-ef0123456789";
+ private static final String GLOBAL_URI = "/nifi-api/process-groups/" +
PROCESS_GROUP_ID + "/download";
+ private static final String CONNECTOR_URI = "/nifi-api/connectors/" +
CONNECTOR_ID + "/process-groups/" + PROCESS_GROUP_ID + "/download";
+
+ @Test
+ public void testCanHandle() {
+ final ExportProcessGroupEndpointMerger merger = new
ExportProcessGroupEndpointMerger();
+
+ assertTrue(merger.canHandle(URI.create(GLOBAL_URI), "GET"));
+
assertTrue(merger.canHandle(URI.create("/nifi-api/process-groups/root/download"),
"GET"));
+ assertTrue(merger.canHandle(URI.create(CONNECTOR_URI), "GET"));
+ assertTrue(merger.canHandle(URI.create(CONNECTOR_URI +
"?includeComponentState=true"), "get"));
+
+ assertFalse(merger.canHandle(URI.create(CONNECTOR_URI), "POST"));
+
assertFalse(merger.canHandle(URI.create("/nifi-api/connectors/not-a-uuid/process-groups/"
+ PROCESS_GROUP_ID + "/download"), "GET"));
+ assertFalse(merger.canHandle(URI.create("/nifi-api/connectors/" +
CONNECTOR_ID + "/process-groups/not-a-uuid/download"), "GET"));
+ assertFalse(merger.canHandle(URI.create(CONNECTOR_URI + "/status"),
"GET"));
+ }
+
+ @Test
+ public void testMergeLocalNodeStates() {
+ final ExportProcessGroupEndpointMerger merger = new
ExportProcessGroupEndpointMerger();
+ final NodeResponse firstNodeResponse = createNodeResponse(1,
createSnapshot(0, "first"));
+ final NodeResponse secondNodeResponse = createNodeResponse(2,
createSnapshot(1, "second"));
+
+ final NodeResponse mergedResponse = merger.merge(
+ URI.create(CONNECTOR_URI), "GET", Set.of(firstNodeResponse,
secondNodeResponse), Collections.emptySet(), firstNodeResponse);
+
+ final ExportedFlowSnapshotEntity mergedEntity =
(ExportedFlowSnapshotEntity) mergedResponse.getUpdatedEntity();
+ final List<VersionedNodeState> localNodeStates =
mergedEntity.getSnapshot().getFlowContents()
+
.getProcessors().iterator().next().getComponentState().getLocalNodeStates();
+ assertEquals(2, localNodeStates.size());
+ assertEquals("first", localNodeStates.get(0).getState().get("value"));
+ assertEquals("second", localNodeStates.get(1).getState().get("value"));
+ }
+
+ private NodeResponse createNodeResponse(final int nodeNumber, final
RegisteredFlowSnapshot flowSnapshot) {
+ final Response response = mock(Response.class);
+ when(response.getStatus()).thenReturn(200);
+
when(response.readEntity(RegisteredFlowSnapshot.class)).thenReturn(flowSnapshot);
+ final NodeIdentifier nodeIdentifier = new NodeIdentifier(
+ "node-" + nodeNumber, "localhost", 8080 + nodeNumber,
"localhost", 9080 + nodeNumber, null, null, null, false);
+ return new NodeResponse(nodeIdentifier, "GET",
URI.create(CONNECTOR_URI), response, 0L, "request");
+ }
+
+ private RegisteredFlowSnapshot createSnapshot(final int nodeOrdinal, final
String value) {
+ final List<VersionedNodeState> localNodeStates = new ArrayList<>();
+ for (int i = 0; i < nodeOrdinal; i++) {
+ localNodeStates.add(null);
+ }
+ localNodeStates.add(new VersionedNodeState(Map.of("value", value)));
+
+ final VersionedComponentState componentState = new
VersionedComponentState();
+ componentState.setLocalNodeStates(localNodeStates);
+ final VersionedProcessor processor = new VersionedProcessor();
+ processor.setIdentifier("processor");
+ processor.setComponentState(componentState);
+ final VersionedProcessGroup processGroup = new VersionedProcessGroup();
+ processGroup.setIdentifier(PROCESS_GROUP_ID);
+ processGroup.setProcessors(Set.of(processor));
+ final RegisteredFlowSnapshot flowSnapshot = new
RegisteredFlowSnapshot();
+ flowSnapshot.setFlowContents(processGroup);
+ return flowSnapshot;
+ }
+}
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 e2c9aed520b..f862c712cbc 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
@@ -2070,6 +2070,19 @@ public interface NiFiServiceFacade {
*/
RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(String
processGroupId, boolean includeReferencedServices, boolean
includeComponentState, boolean mapAssetReferences);
+ /**
+ * Get the current state of a Process Group within a Connector's managed
flow, converted to a Versioned Flow
+ * Snapshot for download.
+ *
+ * @param connectorId the ID of the Connector
+ * @param processGroupId the ID of the Process Group within the
Connector's managed flow
+ * @param includeReferencedServices whether to include referenced
controller services from parent groups
+ * @param includeComponentState whether to include component state in the
export
+ * @return the current Process Group converted to a Versioned Flow
Snapshot for download
+ */
+ RegisteredFlowSnapshot getConnectorCurrentFlowSnapshot(
+ String connectorId, String processGroupId, boolean
includeReferencedServices, boolean includeComponentState);
+
/**
* Returns the name of the Flow Registry that is registered with the given
ID. If no Flow Registry exists with the given ID, will return
* the ID itself as the name
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 56a1d876793..cfe185bad5e 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
@@ -6434,12 +6434,23 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
@Override
public RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(final String
processGroupId, final boolean includeReferencedServices, final boolean
includeComponentState,
final boolean mapAssetReferences) {
+ final ProcessGroup processGroup =
processGroupDAO.getProcessGroup(processGroupId);
+ return getCurrentFlowSnapshot(processGroup, includeReferencedServices,
includeComponentState, mapAssetReferences);
+ }
+
+ @Override
+ public RegisteredFlowSnapshot getConnectorCurrentFlowSnapshot(final String
connectorId, final String processGroupId,
+ final boolean includeReferencedServices, final boolean
includeComponentState) {
+ final ProcessGroup processGroup =
locateConnectorProcessGroup(connectorId, processGroupId);
+ return getCurrentFlowSnapshot(processGroup, includeReferencedServices,
includeComponentState, false);
+ }
+
+ private RegisteredFlowSnapshot getCurrentFlowSnapshot(final ProcessGroup
processGroup, final boolean includeReferencedServices,
+ final boolean includeComponentState, final boolean
mapAssetReferences) {
if (!includeComponentState) {
- return getCurrentFlowSnapshotByGroupId(processGroupId,
includeReferencedServices);
+ return buildFlowSnapshot(processGroup,
processGroup.getIdentifier(), includeReferencedServices, null);
}
- final ProcessGroup processGroup =
processGroupDAO.getProcessGroup(processGroupId);
-
// Validate all processors are stopped and all controller services are
disabled
final List<ProcessorNode> runningProcessors =
processGroup.findAllProcessors().stream()
.filter(p -> p.getPhysicalScheduledState() !=
ScheduledState.STOPPED && p.getPhysicalScheduledState() !=
ScheduledState.DISABLED)
@@ -6474,7 +6485,7 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
.localNodeOrdinal(clusterTopologyProvider.getLocalNodeOrdinal())
.build();
- return buildFlowSnapshot(processGroup, processGroupId,
includeReferencedServices, mappingOptions);
+ return buildFlowSnapshot(processGroup, processGroup.getIdentifier(),
includeReferencedServices, mappingOptions);
}
private Set<String> getAllSubGroups(ProcessGroup processGroup) {
@@ -6487,7 +6498,7 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
private RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(final
String processGroupId, final boolean includeReferencedControllerServices) {
final ProcessGroup processGroup =
processGroupDAO.getProcessGroup(processGroupId);
- return buildFlowSnapshot(processGroup, processGroupId,
includeReferencedControllerServices, null);
+ return getCurrentFlowSnapshot(processGroup,
includeReferencedControllerServices, false, false);
}
private RegisteredFlowSnapshot buildFlowSnapshot(final ProcessGroup
processGroup, final String processGroupId,
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java
index 160a89c2607..fe386e57441 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectorResource.java
@@ -2380,6 +2380,68 @@ public class ConnectorResource extends
ApplicationResource {
return generateOkResponse(entity).build();
}
+ /**
+ * Gets a Process Group within a Connector as a flow definition download.
Available regardless of whether the
+ * Connector is in Troubleshooting mode.
+ *
+ * @param connectorId the connector id
+ * @param processGroupId the process group id within the connector's
managed flow
+ * @param includeReferencedServices whether referenced services from
outside the target group should be included
+ * @param includeComponentState whether component state should be included
+ * @return the flow definition
+ */
+ @GET
+ @Consumes(MediaType.WILDCARD)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{id}/process-groups/{processGroupId}/download")
+ @Operation(
+ summary = "Gets a Process Group within a Connector for download",
+ responses = {
+ @ApiResponse(responseCode = "200", content =
@Content(schema = @Schema(implementation = String.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 - /connectors/{uuid}"),
+ @SecurityRequirement(name = "Write - /connectors/{uuid} -
Only required when includeComponentState is true")
+ },
+ description = "Returns a flow definition for the specified Process
Group within the Connector's managed flow. "
+ + "Available regardless of whether the Connector is in
Troubleshooting mode."
+ )
+ public Response exportConnectorProcessGroup(
+ @Parameter(description = "The connector id.", required = true)
+ @PathParam("id") final String connectorId,
+ @Parameter(description = "The process group id.", required = true)
+ @PathParam("processGroupId") final String processGroupId,
+ @Parameter(description = "Whether referenced services from outside
the target group should be included")
+ @QueryParam("includeReferencedServices")
+ @DefaultValue("false") final boolean includeReferencedServices,
+ @Parameter(description = "Whether component state should be
included. Requires all processors to be stopped and all controller services to
be disabled.")
+ @QueryParam("includeComponentState")
+ @DefaultValue("false") final boolean includeComponentState) {
+
+ if (includeComponentState && isReplicateRequest()) {
+ return replicate(HttpMethod.GET);
+ }
+
+ final RequestAction requiredAction = includeComponentState ?
RequestAction.WRITE : RequestAction.READ;
+ serviceFacade.authorizeAccess(lookup -> {
+ final Authorizable connector = lookup.getConnector(connectorId);
+ connector.authorize(authorizer, requiredAction,
NiFiUserUtils.getNiFiUser());
+ });
+
+ final RegisteredFlowSnapshot flowSnapshot =
serviceFacade.getConnectorCurrentFlowSnapshot(
+ connectorId, processGroupId, includeReferencedServices,
includeComponentState);
+ final String filename =
flowSnapshot.getFlowContents().getName().replaceAll("\\s", "_") + ".json";
+
+ return generateOkResponse(flowSnapshot)
+ .header(HttpHeaders.CONTENT_DISPOSITION,
String.format("attachment; filename=\"%s\"", filename))
+ .build();
+ }
+
/**
* Gets the status history for a Process Group inside the Connector's
managed flow. The status history is available
* regardless of whether the Connector is in Troubleshooting mode.
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 1290d210b10..95f3c6a599b 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
@@ -72,6 +72,7 @@ import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.flow.FlowManager;
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.flow.ExternalControllerServiceReference;
@@ -797,6 +798,92 @@ public class StandardNiFiServiceFacadeTest {
assertNull(versionedFlowSnapshot.getSnapshotMetadata());
}
+ @Test
+ public void testGetConnectorCurrentFlowSnapshot() {
+ final String connectorId = UUID.randomUUID().toString();
+ final String groupId = UUID.randomUUID().toString();
+ final ConnectorDAO connectorDAO = mock(ConnectorDAO.class);
+ serviceFacade.setConnectorDAO(connectorDAO);
+
+ final ConnectorNode connectorNode = mock(ConnectorNode.class);
+ final FrameworkFlowContext flowContext =
mock(FrameworkFlowContext.class);
+ final ProcessGroup managedGroup = mock(ProcessGroup.class);
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(connectorDAO.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(connectorNode);
+ when(connectorNode.getActiveFlowContext()).thenReturn(flowContext);
+ when(flowContext.getManagedProcessGroup()).thenReturn(managedGroup);
+ when(managedGroup.findProcessGroup(groupId)).thenReturn(processGroup);
+ when(processGroup.getIdentifier()).thenReturn(groupId);
+
+ final ExtensionManager extensionManager = mock(ExtensionManager.class);
+
when(flowController.getExtensionManager()).thenReturn(extensionManager);
+ final ControllerServiceProvider controllerServiceProvider =
mock(ControllerServiceProvider.class);
+
when(flowController.getControllerServiceProvider()).thenReturn(controllerServiceProvider);
+
+ final StandardNiFiServiceFacade serviceFacadeSpy = spy(serviceFacade);
+ final VersionedComponentFlowMapper flowMapper =
mock(VersionedComponentFlowMapper.class);
+
when(serviceFacadeSpy.makeNiFiRegistryFlowMapper(extensionManager)).thenReturn(flowMapper);
+ final InstantiatedVersionedProcessGroup nonVersionedProcessGroup =
mock(InstantiatedVersionedProcessGroup.class);
+ when(flowMapper.mapNonVersionedProcessGroup(processGroup,
controllerServiceProvider)).thenReturn(nonVersionedProcessGroup);
+
+ final RegisteredFlowSnapshot flowSnapshot =
+ serviceFacadeSpy.getConnectorCurrentFlowSnapshot(connectorId,
groupId, false, false);
+
+ assertEquals(nonVersionedProcessGroup, flowSnapshot.getFlowContents());
+ verify(connectorDAO).getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY);
+ verify(processGroupDAO, never()).getProcessGroup(anyString());
+ }
+
+ @Test
+ public void
testGetConnectorCurrentFlowSnapshotRejectsProcessGroupOutsideConnector() {
+ final String connectorId = UUID.randomUUID().toString();
+ final String groupId = UUID.randomUUID().toString();
+ final ConnectorDAO connectorDAO = mock(ConnectorDAO.class);
+ serviceFacade.setConnectorDAO(connectorDAO);
+
+ final ConnectorNode connectorNode = mock(ConnectorNode.class);
+ final FrameworkFlowContext flowContext =
mock(FrameworkFlowContext.class);
+ final ProcessGroup managedGroup = mock(ProcessGroup.class);
+ when(connectorDAO.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(connectorNode);
+ when(connectorNode.getActiveFlowContext()).thenReturn(flowContext);
+ when(flowContext.getManagedProcessGroup()).thenReturn(managedGroup);
+
+ assertThrows(ResourceNotFoundException.class,
+ () ->
serviceFacade.getConnectorCurrentFlowSnapshot(connectorId, groupId, false,
false));
+ verify(processGroupDAO, never()).getProcessGroup(anyString());
+ }
+
+ @Test
+ public void testGetConnectorCurrentFlowSnapshotValidatesComponentState() {
+ final String connectorId = UUID.randomUUID().toString();
+ final String groupId = UUID.randomUUID().toString();
+ final ConnectorDAO connectorDAO = mock(ConnectorDAO.class);
+ serviceFacade.setConnectorDAO(connectorDAO);
+
+ final ConnectorNode connectorNode = mock(ConnectorNode.class);
+ final FrameworkFlowContext flowContext =
mock(FrameworkFlowContext.class);
+ final ProcessGroup managedGroup = mock(ProcessGroup.class);
+ final ProcessGroup processGroup = mock(ProcessGroup.class);
+ when(connectorDAO.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(connectorNode);
+ when(connectorNode.getActiveFlowContext()).thenReturn(flowContext);
+ when(flowContext.getManagedProcessGroup()).thenReturn(managedGroup);
+ when(managedGroup.findProcessGroup(groupId)).thenReturn(processGroup);
+
+ final ProcessorNode processor = mock(ProcessorNode.class);
+
when(processor.getPhysicalScheduledState()).thenReturn(ScheduledState.RUNNING,
ScheduledState.STOPPED);
+ when(processGroup.findAllProcessors()).thenReturn(List.of(processor));
+
+ assertThrows(IllegalStateException.class,
+ () ->
serviceFacade.getConnectorCurrentFlowSnapshot(connectorId, groupId, false,
true));
+
+ final ControllerServiceNode controllerService =
mock(ControllerServiceNode.class);
+
when(controllerService.getState()).thenReturn(ControllerServiceState.ENABLED);
+
when(processGroup.findAllControllerServices()).thenReturn(Set.of(controllerService));
+
+ assertThrows(IllegalStateException.class,
+ () ->
serviceFacade.getConnectorCurrentFlowSnapshot(connectorId, groupId, false,
true));
+ }
+
@Test
public void
testGetCurrentFlowSnapshotMapsAssetReferencesOnlyWhenRequested() {
// Asset references are NiFi-internal identifiers that are meaningful
only on this instance, so the general
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestConnectorResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestConnectorResource.java
index c5aeb7fcd90..50ff6617900 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestConnectorResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestConnectorResource.java
@@ -34,6 +34,8 @@ import org.apache.nifi.authorization.user.NiFiUserDetails;
import org.apache.nifi.authorization.user.StandardNiFiUser;
import org.apache.nifi.components.BacklogReportingException;
import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
@@ -1182,6 +1184,54 @@ public class TestConnectorResource {
verify(serviceFacade, never()).getConnectorProcessGroup(anyString(),
anyString());
}
+ @Test
+ public void testExportConnectorProcessGroup() {
+ final AuthorizableLookup lookup = wireMigrationAuthorization();
+ final Authorizable connector = lookup.getConnector(CONNECTOR_ID);
+ final VersionedProcessGroup processGroup = new VersionedProcessGroup();
+ processGroup.setName("Test Process Group");
+ final RegisteredFlowSnapshot flowSnapshot = new
RegisteredFlowSnapshot();
+ flowSnapshot.setFlowContents(processGroup);
+ when(serviceFacade.getConnectorCurrentFlowSnapshot(CONNECTOR_ID,
PROCESS_GROUP_ID, true, false)).thenReturn(flowSnapshot);
+
+ try (Response response =
connectorResource.exportConnectorProcessGroup(CONNECTOR_ID, PROCESS_GROUP_ID,
true, false)) {
+ assertEquals(200, response.getStatus());
+ assertEquals(flowSnapshot, response.getEntity());
+ assertEquals("attachment; filename=\"Test_Process_Group.json\"",
response.getHeaderString("Content-Disposition"));
+ }
+
+ verify(connector).authorize(eq(authorizer), eq(RequestAction.READ),
any());
+ verify(serviceFacade).getConnectorCurrentFlowSnapshot(CONNECTOR_ID,
PROCESS_GROUP_ID, true, false);
+ }
+
+ @Test
+ public void
testExportConnectorProcessGroupWithComponentStateRequiresWrite() {
+ final AuthorizableLookup lookup = wireMigrationAuthorization();
+ final Authorizable connector = lookup.getConnector(CONNECTOR_ID);
+ final VersionedProcessGroup processGroup = new VersionedProcessGroup();
+ processGroup.setName("Test Process Group");
+ final RegisteredFlowSnapshot flowSnapshot = new
RegisteredFlowSnapshot();
+ flowSnapshot.setFlowContents(processGroup);
+ when(serviceFacade.getConnectorCurrentFlowSnapshot(CONNECTOR_ID,
PROCESS_GROUP_ID, false, true)).thenReturn(flowSnapshot);
+
+ try (Response response =
connectorResource.exportConnectorProcessGroup(CONNECTOR_ID, PROCESS_GROUP_ID,
false, true)) {
+ assertEquals(200, response.getStatus());
+ }
+
+ verify(connector).authorize(eq(authorizer), eq(RequestAction.WRITE),
any());
+ verify(serviceFacade).getConnectorCurrentFlowSnapshot(CONNECTOR_ID,
PROCESS_GROUP_ID, false, true);
+ }
+
+ @Test
+ public void testExportConnectorProcessGroupNotAuthorized() {
+
doThrow(AccessDeniedException.class).when(serviceFacade).authorizeAccess(any(AuthorizeAccess.class));
+
+ assertThrows(AccessDeniedException.class, () ->
+ connectorResource.exportConnectorProcessGroup(CONNECTOR_ID,
PROCESS_GROUP_ID, false, false));
+
+ verify(serviceFacade,
never()).getConnectorCurrentFlowSnapshot(anyString(), anyString(), eq(false),
eq(false));
+ }
+
@Test
public void testGetConnectorControllerServiceState() {
final ComponentStateDTO stateDTO = new ComponentStateDTO();
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.spec.ts
new file mode 100644
index 00000000000..361eefc1958
--- /dev/null
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.spec.ts
@@ -0,0 +1,80 @@
+/*
+ * 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 { provideHttpClient } from '@angular/common/http';
+import { TestBed } from '@angular/core/testing';
+import { UnsafeApiPathError } from '@nifi/shared';
+import { Client } from '../../../service/client.service';
+import { ClusterConnectionService } from
'../../../service/cluster-connection.service';
+import { ConnectorService } from './connector.service';
+
+describe('ConnectorService', () => {
+ async function setup() {
+ await TestBed.configureTestingModule({
+ providers: [
+ ConnectorService,
+ provideHttpClient(),
+ { provide: Client, useValue: {} },
+ { provide: ClusterConnectionService, useValue: {} }
+ ]
+ }).compileComponents();
+
+ return {
+ service: TestBed.inject(ConnectorService),
+ open: vi.spyOn(window, 'open').mockImplementation(() => null)
+ };
+ }
+
+ afterEach(() => {
+ vi.restoreAllMocks();
+ });
+
+ it('should encode Connector and Process Group identifiers in the download
URL', async () => {
+ const { service, open } = await setup();
+
+ service.downloadFlow({
+ connectorId: 'connector id',
+ processGroupId: 'process?group',
+ includeReferencedServices: true
+ });
+
+ expect(open).toHaveBeenCalledWith(
+
'../nifi-api/connectors/connector%20id/process-groups/process%3Fgroup/download?includeReferencedServices=true',
+ '_blank',
+ 'noreferrer'
+ );
+ });
+
+ it.each([
+ ['Connector', '..%2F..%2Fnifi-api%2Fcontroller', 'process-group'],
+ ['Process Group', 'connector', '../../controller']
+ ])(
+ 'should reject a path-traversal %s identifier without opening a
window',
+ async (_segment, connectorId, processGroupId) => {
+ const { service, open } = await setup();
+
+ expect(() =>
+ service.downloadFlow({
+ connectorId,
+ processGroupId,
+ includeReferencedServices: false
+ })
+ ).toThrow(UnsafeApiPathError);
+ expect(open).not.toHaveBeenCalled();
+ }
+ );
+});
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.ts
index 3ffba043cd5..19f74d42c60 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/service/connector.service.ts
@@ -22,7 +22,7 @@ import { HttpClient } from '@angular/common/http';
import { Client } from '../../../service/client.service';
import { ClusterConnectionService } from
'../../../service/cluster-connection.service';
import { ChangeConnectorVersionRequest, ConnectorsResponse,
CreateConnectorRequest } from '../state';
-import { ConnectorEntity, ProcessGroupEntity } from '@nifi/shared';
+import { ConnectorEntity, ProcessGroupEntity, safeApiPath } from
'@nifi/shared';
import {
ControllerServiceEntity,
ParameterContextEntity,
@@ -30,6 +30,7 @@ import {
SearchResultsEntity
} from '../../../state/shared';
import { DropRequestEntity } from '../../../state/empty-queue';
+import type { DownloadConnectorFlowRequest } from
'../state/connector-canvas/connector-canvas.actions';
@Injectable({ providedIn: 'root' })
export class ConnectorService {
@@ -138,6 +139,14 @@ export class ConnectorService {
);
}
+ downloadFlow(request: DownloadConnectorFlowRequest): void {
+ window.open(
+
`${ConnectorService.API}/connectors/${safeApiPath(request.connectorId)}/process-groups/${safeApiPath(request.processGroupId)}/download?includeReferencedServices=${request.includeReferencedServices}`,
+ '_blank',
+ 'noreferrer'
+ );
+ }
+
getConnectorControllerServices(connectorId: string, processGroupId:
string): Observable<any> {
return this.httpClient.get(
`${ConnectorService.API}/connectors/${connectorId}/flow/process-groups/${processGroupId}/controller-services`
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
index 48f7f107d6e..20bb8331aac 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.actions.ts
@@ -58,6 +58,12 @@ export interface SelectComponentsRequest {
components: SelectedComponent[];
}
+export interface DownloadConnectorFlowRequest {
+ connectorId: string;
+ processGroupId: string;
+ includeReferencedServices: boolean;
+}
+
export const loadConnectorFlow = createAction(
'[Connector Canvas] Load Connector Flow',
props<{ connectorId: string; processGroupId: string }>()
@@ -416,4 +422,9 @@ export const moveCanvasComponentsFailure = createAction(
props<{ components: MoveCanvasComponentRef[]; error: string }>()
);
+export const downloadFlow = createAction(
+ '[Connector Canvas] Download Flow',
+ props<{ request: DownloadConnectorFlowRequest }>()
+);
+
export const resetConnectorCanvasState = createAction('[Connector Canvas]
Reset State');
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
index 527a9df881d..64561fb26b7 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.spec.ts
@@ -59,6 +59,7 @@ import {
createConnectionFailure,
createConnectionRequested,
createConnectionSuccess,
+ downloadFlow,
enterProcessGroup,
leaveProcessGroup,
loadConnectorFlow,
@@ -146,6 +147,7 @@ describe('ConnectorCanvasEffects', () => {
// Mock services
const mockConnectorService = {
+ downloadFlow: vi.fn(),
getConnectorFlow: vi.fn(),
getProcessGroup: vi.fn(),
getConnectorParameterContext: vi.fn(),
@@ -717,6 +719,25 @@ describe('ConnectorCanvasEffects', () => {
});
});
+ describe('downloadFlow$', () => {
+ it('should delegate the complete request to ConnectorService without
dispatching another action', async () => {
+ const { effects, actions$, mockConnectorService } = await setup();
+ const action = downloadFlow({
+ request: {
+ connectorId: 'connector-1',
+ processGroupId: 'process-group-1',
+ includeReferencedServices: true
+ }
+ });
+ actions$(of(action));
+
+ const result = await firstValueFrom(effects.downloadFlow$);
+
+
expect(mockConnectorService.downloadFlow).toHaveBeenCalledWith(action.request);
+ expect(result).toBe(action);
+ });
+ });
+
describe('enterProcessGroup$', () => {
it('should navigate to the child process group canvas URL', async ()
=> {
const { effects, actions$, mockRouter } = await setup({
connectorId: 'my-connector' });
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
index b1609dd5ce6..af00b696be2 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/state/connector-canvas/connector-canvas.effects.ts
@@ -239,6 +239,15 @@ export class ConnectorCanvasEffects {
)
);
+ downloadFlow$ = createEffect(
+ () =>
+ this.actions$.pipe(
+ ofType(ConnectorCanvasActions.downloadFlow),
+ tap(({ request }) =>
this.connectorService.downloadFlow(request))
+ ),
+ { dispatch: false }
+ );
+
loadConnectorFlowComplete$ = createEffect(() =>
this.actions$.pipe(
ofType(ConnectorCanvasActions.loadConnectorFlowSuccess),
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
index 8502f4384a1..6e2ac199874 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.spec.ts
@@ -53,6 +53,7 @@ import {
import { selectParentProcessGroupId } from
'../../state/connector-canvas/connector-canvas.selectors';
import {
deselectAllComponents,
+ downloadFlow,
enterProcessGroup,
leaveProcessGroup,
loadConnectorFlow,
@@ -290,6 +291,7 @@ interface SetupOptions {
connectedToCluster?: boolean;
contentViewerUrl?: string | null;
connections?: ReturnType<typeof makeConnection>[];
+ connectorCanRead?: boolean;
}
const DEFAULT_CONNECTOR_ID = 'connector-1';
@@ -348,7 +350,29 @@ function buildMockSelectors(options: SetupOptions = {}) {
{ selector: selectUrl, value: url },
{ selector: selectRouteParams, value: routeParamsValue },
{ selector: selectCurrentUser, value:
buildMockCurrentUser(canAccessProvenance) },
- { selector: selectConnectorCanvasEntity, value: null },
+ {
+ selector: selectConnectorCanvasEntity,
+ value:
+ options.connectorCanRead === undefined
+ ? null
+ : {
+ id: DEFAULT_CONNECTOR_ID,
+ uri: `/connectors/${DEFAULT_CONNECTOR_ID}`,
+ revision: { version: 1 },
+ permissions: { canRead: options.connectorCanRead,
canWrite: false },
+ bulletins: [],
+ status: {} as never,
+ component: {
+ id: DEFAULT_CONNECTOR_ID,
+ name: 'Connector',
+ type: 'type',
+ state: 'STOPPED',
+ bundle: { group: 'group', artifact: 'artifact',
version: '1' },
+ availableActions: [],
+ managedProcessGroupId: DEFAULT_PROCESS_GROUP_ID
+ }
+ }
+ },
{ selector: selectConnectorCanvasEntitySaving, value: false },
{ selector: selectConnectorProvenanceEvents, value:
options.provenanceEvents ?? [] },
{ selector: selectConnectorProvenanceStatus, value:
options.provenanceStatus ?? 'pending' },
@@ -1086,6 +1110,17 @@ describe('ConnectorCanvasComponent', () => {
expect(result).toBeUndefined();
});
+ it('should return download menu variants', () => {
+ const { component } = setup();
+ const menu = component.contextMenuProvider.getMenu('download');
+
+ expect(menu).toBeDefined();
+ expect(menu!.menuItems.map((item) => item.text)).toEqual([
+ 'Without External Services',
+ 'With External Services'
+ ]);
+ });
+
it('should return canvas menu with Refresh, Leave Group, and Empty
All Queues when targetType is canvas', fakeAsync(() => {
const { fixture, component } = setup({ parentProcessGroupId:
'parent-pg' });
fixture.detectChanges();
@@ -1101,6 +1136,7 @@ describe('ConnectorCanvasComponent', () => {
'Refresh',
'Leave Group',
'Controller Services',
+ 'Download Flow Definition',
'Empty All Queues',
'Drain',
'Cancel Drain'
@@ -1127,6 +1163,7 @@ describe('ConnectorCanvasComponent', () => {
'Empty Queue',
'Empty All Queues',
'Controller Services',
+ 'Download Flow Definition',
'Center In View'
]);
}));
@@ -1139,6 +1176,106 @@ describe('ConnectorCanvasComponent', () => {
});
});
+ describe('download menu', () => {
+ it('should show Download Flow Definition on the canvas when the
Connector is readable', fakeAsync(() => {
+ const { fixture, component } = setup({ connectorCanRead: true
});
+ fixture.detectChanges();
+ tick();
+
+ component.onContextMenuOpened(buildCanvasContext());
+ const menu = component.contextMenuProvider.getMenu('root')!;
+ const download = menu.menuItems.find((item) => item.text ===
'Download Flow Definition');
+
+ expect(download).toBeDefined();
+ expect(download!.condition!(null)).toBe(true);
+ }));
+
+ it('should hide Download Flow Definition when the Connector is not
readable', fakeAsync(() => {
+ const { fixture, component } = setup({ connectorCanRead: false
});
+ fixture.detectChanges();
+ tick();
+
+ component.onContextMenuOpened(buildCanvasContext());
+ const menu = component.contextMenuProvider.getMenu('root')!;
+ const download = menu.menuItems.find((item) => item.text ===
'Download Flow Definition');
+
+ expect(download).toBeDefined();
+ expect(download!.condition!(null)).toBe(false);
+ }));
+
+ it('should show Download Flow Definition for a single readable
Process Group', fakeAsync(() => {
+ const { fixture, component } = setup({ connectorCanRead: true
});
+ fixture.detectChanges();
+ tick();
+
+
component.onContextMenuOpened(buildComponentContext(ComponentType.ProcessGroup,
'pg-child'));
+ const menu = component.contextMenuProvider.getMenu('root')!;
+ const download = menu.menuItems.find((item) => item.text ===
'Download Flow Definition');
+
+ expect(download).toBeDefined();
+ expect(download!.condition!(null)).toBe(true);
+ }));
+
+ it('should show Download Flow Definition for an unreadable Process
Group when the Connector is readable', fakeAsync(() => {
+ const { fixture, component } = setup({ connectorCanRead: true
});
+ fixture.detectChanges();
+ tick();
+
+ component.onContextMenuOpened(
+ buildComponentContext(ComponentType.ProcessGroup,
'pg-child', 1, {
+ permissions: { canRead: false, canWrite: false }
+ })
+ );
+ const menu = component.contextMenuProvider.getMenu('root')!;
+ const download = menu.menuItems.find((item) => item.text ===
'Download Flow Definition');
+
+ expect(download).toBeDefined();
+ expect(download!.condition!(null)).toBe(true);
+ }));
+
+ it('should dispatch download for the current Process Group without
external services', fakeAsync(() => {
+ const { fixture, component, dispatchSpy } = setup({
connectorCanRead: true });
+ fixture.detectChanges();
+ tick();
+ dispatchSpy.mockClear();
+
+ component.onContextMenuOpened(buildCanvasContext());
+ const menu =
component.contextMenuProvider.getMenu('download')!;
+ menu.menuItems.find((item) => item.text === 'Without External
Services')!.action!(null);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(
+ downloadFlow({
+ request: {
+ connectorId: DEFAULT_CONNECTOR_ID,
+ processGroupId: DEFAULT_PROCESS_GROUP_ID,
+ includeReferencedServices: false
+ }
+ })
+ );
+ }));
+
+ it('should dispatch download for the selected Process Group with
external services', fakeAsync(() => {
+ const { fixture, component, dispatchSpy } = setup({
connectorCanRead: true });
+ fixture.detectChanges();
+ tick();
+ dispatchSpy.mockClear();
+
+
component.onContextMenuOpened(buildComponentContext(ComponentType.ProcessGroup,
'pg-child'));
+ const menu =
component.contextMenuProvider.getMenu('download')!;
+ menu.menuItems.find((item) => item.text === 'With External
Services')!.action!(null);
+
+ expect(dispatchSpy).toHaveBeenCalledWith(
+ downloadFlow({
+ request: {
+ connectorId: DEFAULT_CONNECTOR_ID,
+ processGroupId: 'pg-child',
+ includeReferencedServices: true
+ }
+ })
+ );
+ }));
+ });
+
describe('canvas menu conditions', () => {
it('should always show Refresh', fakeAsync(() => {
const { fixture, component } = setup();
diff --git
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
index ea9778d43c1..617f4962494 100644
---
a/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
+++
b/nifi-frontend/src/main/frontend/apps/nifi/src/app/pages/connectors/ui/connector-canvas/connector-canvas.component.ts
@@ -214,6 +214,24 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
contextMenuProvider: ContextMenuDefinitionProvider = {
getMenu: (menuId: string): ContextMenuDefinition | undefined => {
+ if (menuId === 'download') {
+ return {
+ id: menuId,
+ menuItems: [
+ {
+ text: 'Without External Services',
+ clazz: 'fa',
+ action: () => this.downloadFlowAction(false)
+ },
+ {
+ text: 'With External Services',
+ clazz: 'fa',
+ action: () => this.downloadFlowAction(true)
+ }
+ ]
+ };
+ }
+
if (menuId !== 'root') {
return undefined;
}
@@ -244,6 +262,12 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
condition: () => !!this.currentProcessGroupId,
action: () =>
this.controllerServicesAction(this.currentProcessGroupId!)
},
+ {
+ text: 'Download Flow Definition',
+ clazz: 'fa',
+ subMenuId: 'download',
+ condition: () => this.canDownloadFlow()
+ },
{ isSeparator: true },
{
text: 'Empty All Queues',
@@ -338,6 +362,12 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
condition: () => isProcessGroup && isSingleSelection,
action: () =>
this.controllerServicesAction(clicked!.entity.id)
},
+ {
+ text: 'Download Flow Definition',
+ clazz: 'fa',
+ subMenuId: 'download',
+ condition: () => this.canDownloadFlow()
+ },
{ isSeparator: true },
this.getCenterInViewMenuItem()
];
@@ -1039,6 +1069,41 @@ export class ConnectorCanvasComponent implements OnInit,
OnDestroy {
this.store.dispatch(ConnectorCanvasActions.navigateToControllerServices({
processGroupId }));
}
+ downloadFlowAction(includeReferencedServices: boolean): void {
+ const processGroupId = this.getDownloadProcessGroupId();
+ if (this.currentConnectorId && processGroupId) {
+ this.store.dispatch(
+ ConnectorCanvasActions.downloadFlow({
+ request: {
+ connectorId: this.currentConnectorId,
+ processGroupId,
+ includeReferencedServices
+ }
+ })
+ );
+ }
+ }
+
+ private canDownloadFlow(): boolean {
+ return this.connectorEntity()?.permissions.canRead === true &&
this.getDownloadProcessGroupId() !== null;
+ }
+
+ private getDownloadProcessGroupId(): string | null {
+ const context = this.currentContextMenuContext;
+ if (context?.targetType === 'canvas') {
+ return this.currentProcessGroupId;
+ }
+
+ if (context?.targetType === 'component') {
+ const clicked = context.clickedComponent;
+ if (context.selectedComponents.length <= 1 &&
clicked?.ui.componentType === ComponentType.ProcessGroup) {
+ return clicked.entity.id;
+ }
+ }
+
+ return null;
+ }
+
canDrain(): boolean {
const entity = this.connectorEntity();
if (!entity || this.entitySaving()) {
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java
index 5d60a51feb2..cecc4266440 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ConnectorTroubleshootingIT.java
@@ -289,6 +289,37 @@ public class ConnectorTroubleshootingIT extends
NiFiSystemIT {
getNifiClient().getProcessorClient().deleteProcessor(refreshedForDelete);
}
+ /**
+ * Verifies Connector-scoped flow definition export outside
Troubleshooting for the managed root and a child
+ * Process Group.
+ */
+ @Test
+ public void testExportProcessGroupOutsideTroubleshooting() throws
NiFiClientException, IOException, InterruptedException {
+ final ConnectorEntity connector =
getClientUtil().createConnector("ComponentLifecycleConnector");
+ final String connectorId = connector.getId();
+
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().waitForValidConnector(connectorId);
+ assertConnectorState(connectorId, ConnectorState.STOPPED);
+
+ final String managedGroupId =
getNifiClient().getConnectorClient().getConnector(connectorId).getComponent().getManagedProcessGroupId();
+ final String childGroupId = findFirstChildProcessGroupId(connectorId);
+ assertNotNull(childGroupId, "Managed flow should contain a child
Process Group");
+
+ final File managedGroupExport =
Files.createTempFile("connector-managed-group-", ".json").toFile();
+ final File childGroupExport =
Files.createTempFile("connector-child-group-", ".json").toFile();
+ try {
+
getNifiClient().getConnectorClient().exportProcessGroup(connectorId,
managedGroupId, false, managedGroupExport);
+
getNifiClient().getConnectorClient().exportProcessGroup(connectorId,
childGroupId, true, childGroupExport);
+
+ assertTrue(managedGroupExport.length() > 0);
+ assertTrue(childGroupExport.length() > 0);
+ } finally {
+ Files.deleteIfExists(managedGroupExport.toPath());
+ Files.deleteIfExists(childGroupExport.toPath());
+ }
+ }
+
/**
* Stop a connector, enter Troubleshooting, start processors inside the
managed flow, restart NiFi, and ensure the
* processors are still running and the Connector remains in
Troubleshooting.
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ConnectorClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ConnectorClient.java
index bcf3019d4ee..7946c203de2 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ConnectorClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ConnectorClient.java
@@ -341,6 +341,35 @@ public interface ConnectorClient {
*/
ProcessGroupEntity getProcessGroup(String connectorId, String
processGroupId) throws NiFiClientException, IOException;
+ /**
+ * Exports a Process Group within a Connector's managed flow.
+ *
+ * @param connectorId the connector ID
+ * @param processGroupId the process group ID within the connector's
managed flow
+ * @param includeReferencedServices whether referenced services should be
included
+ * @param outputFile the file to write
+ * @return the output file
+ * @throws NiFiClientException if an error occurs during the request
+ * @throws IOException if an I/O error occurs
+ */
+ File exportProcessGroup(String connectorId, String processGroupId, boolean
includeReferencedServices, File outputFile)
+ throws NiFiClientException, IOException;
+
+ /**
+ * Exports a Process Group within a Connector's managed flow.
+ *
+ * @param connectorId the connector ID
+ * @param processGroupId the process group ID within the connector's
managed flow
+ * @param includeReferencedServices whether referenced services should be
included
+ * @param includeComponentState whether component state should be included
+ * @param outputFile the file to write
+ * @return the output file
+ * @throws NiFiClientException if an error occurs during the request
+ * @throws IOException if an I/O error occurs
+ */
+ File exportProcessGroup(String connectorId, String processGroupId, boolean
includeReferencedServices,
+ boolean includeComponentState, File outputFile) throws
NiFiClientException, IOException;
+
/**
* Gets the status for the process group managed by a connector.
*
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyConnectorClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyConnectorClient.java
index db57054154e..fe63bac6b0c 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyConnectorClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyConnectorClient.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.toolkit.client.impl;
+import jakarta.ws.rs.WebApplicationException;
import jakarta.ws.rs.client.Entity;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.MediaType;
@@ -608,6 +609,49 @@ public class JerseyConnectorClient extends
AbstractJerseyClient implements Conne
});
}
+ @Override
+ public File exportProcessGroup(final String connectorId, final String
processGroupId, final boolean includeReferencedServices,
+ final File outputFile) throws NiFiClientException, IOException {
+ return exportProcessGroup(connectorId, processGroupId,
includeReferencedServices, false, outputFile);
+ }
+
+ @Override
+ public File exportProcessGroup(final String connectorId, final String
processGroupId, final boolean includeReferencedServices,
+ final boolean includeComponentState, final File outputFile) throws
NiFiClientException, IOException {
+ if (StringUtils.isBlank(connectorId)) {
+ throw new IllegalArgumentException("Connector id cannot be null or
blank");
+ }
+ if (StringUtils.isBlank(processGroupId)) {
+ throw new IllegalArgumentException("Process group id cannot be
null or blank");
+ }
+
+ return executeAction("Error exporting process group for Connector " +
connectorId, () -> {
+ final WebTarget target = connectorTarget
+ .path("/process-groups/{processGroupId}/download")
+ .resolveTemplate("id", connectorId)
+ .resolveTemplate("processGroupId", processGroupId)
+ .queryParam("includeReferencedServices",
includeReferencedServices)
+ .queryParam("includeComponentState",
includeComponentState);
+
+ final Response response = getRequestBuilder(target)
+ .accept(MediaType.APPLICATION_JSON)
+ .get();
+ if (response.getStatusInfo().getFamily() !=
Response.Status.Family.SUCCESSFUL) {
+ throw new WebApplicationException(response);
+ }
+
+ final String contentDispositionHeader =
response.getHeaderString("Content-Disposition");
+ if (StringUtils.isBlank(contentDispositionHeader)) {
+ throw new IllegalStateException("Content-Disposition header
was blank or missing");
+ }
+
+ try (final InputStream responseInputStream =
response.readEntity(InputStream.class)) {
+ Files.copy(responseInputStream, outputFile.toPath(),
StandardCopyOption.REPLACE_EXISTING);
+ return outputFile;
+ }
+ });
+ }
+
@Override
public ProcessGroupStatusEntity getStatus(final String connectorId, final
boolean recursive) throws NiFiClientException, IOException {
Objects.requireNonNull(connectorId, "Connector ID required");
