You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2019/07/29 14:29:19 UTC
[nifi] branch master updated: NIFI-5856: When exporting a flow that
references a Controller Service,
also include the ExternalControllerServiceReference's in the snapshot. When
importing/updating a flow,
if a Controller Service that is inherited is referenced,
attempt to find any one ControllerService that gets inherited whose name
matches the reference and if found link to that Controller Service.
This is an automated email from the ASF dual-hosted git repository.
bbende pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new f8e9318 NIFI-5856: When exporting a flow that references a Controller Service, also include the ExternalControllerServiceReference's in the snapshot. When importing/updating a flow, if a Controller Service that is inherited is referenced, attempt to find any one ControllerService that gets inherited whose name matches the reference and if found link to that Controller Service.
f8e9318 is described below
commit f8e93186f53917b1fddbc2ae3de26b65a99b9246
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Tue Jun 25 10:04:25 2019 -0400
NIFI-5856: When exporting a flow that references a Controller Service, also include the ExternalControllerServiceReference's in the snapshot. When importing/updating a flow, if a Controller Service that is inherited is referenced, attempt to find any one ControllerService that gets inherited whose name matches the reference and if found link to that Controller Service.
NIFI-5856: Fixed checkstyle violation
NIFI-5856: Addressed review feedback
NIFI-5856: When assigning a matching Controller Service at a higher level for Process Group import/update, ensure that we consider the User's read permissions when selecting the Controller Service
This closes #3553.
Signed-off-by: Bryan Bende <bb...@apache.org>
---
.../apache/nifi/registry/flow/FlowRegistry.java | 7 +-
.../apache/nifi/groups/StandardProcessGroup.java | 3 +-
.../nifi/registry/flow/RestBasedFlowRegistry.java | 16 +--
.../mapping/InstantiatedVersionedProcessGroup.java | 12 ++
.../flow/mapping/NiFiRegistryFlowMapper.java | 69 +++++++++--
.../java/org/apache/nifi/util/BundleUtils.java | 30 ++---
.../nifi/integration/versioned/ImportFlowIT.java | 5 +-
.../org/apache/nifi/web/NiFiServiceFacade.java | 17 ++-
.../org/apache/nifi/web/NiFiServiceFacadeLock.java | 10 +-
.../apache/nifi/web/StandardNiFiServiceFacade.java | 136 ++++++++++++++++++++-
.../apache/nifi/web/api/ProcessGroupResource.java | 35 ++++--
.../org/apache/nifi/web/api/VersionsResource.java | 9 +-
12 files changed, 289 insertions(+), 60 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/FlowRegistry.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/FlowRegistry.java
index b883274..9821eaf 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/FlowRegistry.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/FlowRegistry.java
@@ -22,6 +22,7 @@ import org.apache.nifi.registry.bucket.Bucket;
import org.apache.nifi.registry.client.NiFiRegistryException;
import java.io.IOException;
+import java.util.Map;
import java.util.Set;
public interface FlowRegistry {
@@ -130,6 +131,8 @@ public interface FlowRegistry {
*
* @param flow the Versioned Flow
* @param snapshot the snapshot of the flow
+ * @param externalControllerServices a mapping of of Controller Service identifier to ExternalControllerServiceReference for any Controller Service that is referenced by the flow but that are
+ * not included as part of the VersionedProcessGroup
* @param comments any comments for the snapshot
* @param expectedVersion the version of the flow that we expect to save this snapshot as
* @return the versioned flow snapshot
@@ -138,8 +141,8 @@ public interface FlowRegistry {
* @throws NullPointerException if the VersionedFlow is null, or if its bucket identifier is null, or if the flow to snapshot is null
* @throws NiFiRegistryException if the flow does not exist
*/
- VersionedFlowSnapshot registerVersionedFlowSnapshot(VersionedFlow flow, VersionedProcessGroup snapshot, String comments, int expectedVersion, NiFiUser user)
- throws IOException, NiFiRegistryException;
+ VersionedFlowSnapshot registerVersionedFlowSnapshot(VersionedFlow flow, VersionedProcessGroup snapshot, Map<String, ExternalControllerServiceReference> externalControllerServices, String comments,
+ int expectedVersion, NiFiUser user) throws IOException, NiFiRegistryException;
/**
* Returns the latest (most recent) version of the Flow in the Flow Registry for the given bucket and flow
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index d443a37..54abd35 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -3371,7 +3371,8 @@ public final class StandardProcessGroup implements ProcessGroup {
if (ComponentType.CONTROLLER_SERVICE == component.getComponentType()) {
final ControllerServiceNode serviceNode = getVersionedControllerService(this, component.getIdentifier());
if (serviceNode != null) {
- final VersionedControllerService versionedService = mapper.mapControllerService(serviceNode, controllerServiceProvider);
+ final VersionedControllerService versionedService = mapper.mapControllerService(serviceNode, controllerServiceProvider,
+ Collections.singleton(serviceNode.getProcessGroupIdentifier()), new HashMap<>());
final Set<FlowDifference> differences = flowComparator.compareControllerServices(versionedService, (VersionedControllerService) component);
if (!differences.isEmpty()) {
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/RestBasedFlowRegistry.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/RestBasedFlowRegistry.java
index 21e5e0c..261c0aa 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/RestBasedFlowRegistry.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/RestBasedFlowRegistry.java
@@ -17,12 +17,6 @@
package org.apache.nifi.registry.flow;
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.Set;
-
-import javax.net.ssl.SSLContext;
-
import org.apache.nifi.authorization.user.NiFiUser;
import org.apache.nifi.registry.bucket.Bucket;
import org.apache.nifi.registry.client.BucketClient;
@@ -33,6 +27,12 @@ import org.apache.nifi.registry.client.NiFiRegistryClientConfig;
import org.apache.nifi.registry.client.NiFiRegistryException;
import org.apache.nifi.registry.client.impl.JerseyNiFiRegistryClient;
+import javax.net.ssl.SSLContext;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
public class RestBasedFlowRegistry implements FlowRegistry {
private final FlowRegistryClient flowRegistryClient;
@@ -175,11 +175,13 @@ public class RestBasedFlowRegistry implements FlowRegistry {
@Override
public VersionedFlowSnapshot registerVersionedFlowSnapshot(final VersionedFlow flow, final VersionedProcessGroup snapshot,
- final String comments, final int expectedVersion, final NiFiUser user) throws IOException, NiFiRegistryException {
+ final Map<String, ExternalControllerServiceReference> externalControllerServices, final String comments, final int expectedVersion,
+ final NiFiUser user) throws IOException, NiFiRegistryException {
final FlowSnapshotClient snapshotClient = getFlowSnapshotClient(user);
final VersionedFlowSnapshot versionedFlowSnapshot = new VersionedFlowSnapshot();
versionedFlowSnapshot.setFlowContents(snapshot);
+ versionedFlowSnapshot.setExternalControllerServices(externalControllerServices);
final VersionedFlowSnapshotMetadata metadata = new VersionedFlowSnapshotMetadata();
metadata.setBucketIdentifier(flow.getBucketIdentifier());
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/InstantiatedVersionedProcessGroup.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/InstantiatedVersionedProcessGroup.java
index a669220..3eed715 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/InstantiatedVersionedProcessGroup.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/InstantiatedVersionedProcessGroup.java
@@ -17,11 +17,15 @@
package org.apache.nifi.registry.flow.mapping;
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
import org.apache.nifi.registry.flow.VersionedProcessGroup;
+import java.util.Map;
+
public class InstantiatedVersionedProcessGroup extends VersionedProcessGroup implements InstantiatedVersionedComponent {
private final String instanceId;
private final String groupId;
+ private Map<String, ExternalControllerServiceReference> externalControllerServiceReferences;
public InstantiatedVersionedProcessGroup(final String instanceId, final String instanceGroupId) {
this.instanceId = instanceId;
@@ -37,4 +41,12 @@ public class InstantiatedVersionedProcessGroup extends VersionedProcessGroup imp
public String getInstanceGroupId() {
return groupId;
}
+
+ public void setExternalControllerServiceReferences(final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences) {
+ this.externalControllerServiceReferences = externalControllerServiceReferences;
+ }
+
+ public Map<String, ExternalControllerServiceReference> getExternalControllerServiceReferences() {
+ return externalControllerServiceReferences;
+ }
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/NiFiRegistryFlowMapper.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/NiFiRegistryFlowMapper.java
index e3aba02..4780381 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/NiFiRegistryFlowMapper.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/registry/flow/mapping/NiFiRegistryFlowMapper.java
@@ -42,6 +42,7 @@ import org.apache.nifi.registry.flow.ComponentType;
import org.apache.nifi.registry.flow.ConnectableComponent;
import org.apache.nifi.registry.flow.ConnectableComponentType;
import org.apache.nifi.registry.flow.ControllerServiceAPI;
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
import org.apache.nifi.registry.flow.FlowRegistry;
import org.apache.nifi.registry.flow.FlowRegistryClient;
import org.apache.nifi.registry.flow.PortType;
@@ -93,8 +94,8 @@ public class NiFiRegistryFlowMapper {
public InstantiatedVersionedProcessGroup mapProcessGroup(final ProcessGroup group, final ControllerServiceProvider serviceProvider, final FlowRegistryClient registryClient,
final boolean mapDescendantVersionedFlows) {
versionedComponentIds.clear();
- final InstantiatedVersionedProcessGroup mapped = mapGroup(group, serviceProvider, registryClient, true, mapDescendantVersionedFlows);
+ final InstantiatedVersionedProcessGroup mapped = mapGroup(group, serviceProvider, registryClient, true, mapDescendantVersionedFlows);
populateReferencedAncestorVariables(group, mapped);
return mapped;
@@ -136,9 +137,22 @@ public class NiFiRegistryFlowMapper {
populateVariableNames(group.getParent(), variableNames);
}
+ private InstantiatedVersionedProcessGroup mapGroup(final ProcessGroup group, final ControllerServiceProvider serviceProvider, final FlowRegistryClient registryClient,
+ final boolean topLevel, final boolean mapDescendantVersionedFlows) {
+
+ final Set<String> allIncludedGroupsIds = group.findAllProcessGroups().stream()
+ .map(ProcessGroup::getIdentifier)
+ .collect(Collectors.toSet());
+ allIncludedGroupsIds.add(group.getIdentifier());
+
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences = new HashMap<>();
+ return mapGroup(group, serviceProvider, registryClient, topLevel, mapDescendantVersionedFlows, allIncludedGroupsIds, externalControllerServiceReferences);
+ }
- private InstantiatedVersionedProcessGroup mapGroup(final ProcessGroup group, final ControllerServiceProvider serviceLookup, final FlowRegistryClient registryClient,
- final boolean topLevel, final boolean mapDescendantVersionedFlows) {
+
+ private InstantiatedVersionedProcessGroup mapGroup(final ProcessGroup group, final ControllerServiceProvider serviceProvider, final FlowRegistryClient registryClient,
+ final boolean topLevel, final boolean mapDescendantVersionedFlows, final Set<String> includedGroupIds,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences) {
final InstantiatedVersionedProcessGroup versionedGroup = new InstantiatedVersionedProcessGroup(group.getIdentifier(), group.getProcessGroupIdentifier());
versionedGroup.setIdentifier(getId(group.getVersionedComponentId(), group.getIdentifier()));
@@ -184,7 +198,7 @@ public class NiFiRegistryFlowMapper {
}
versionedGroup.setControllerServices(group.getControllerServices(false).stream()
- .map(service -> mapControllerService(service, serviceLookup))
+ .map(service -> mapControllerService(service, serviceProvider, includedGroupIds, externalControllerServiceReferences))
.collect(Collectors.toCollection(LinkedHashSet::new)));
versionedGroup.setFunnels(group.getFunnels().stream()
@@ -204,7 +218,7 @@ public class NiFiRegistryFlowMapper {
.collect(Collectors.toCollection(LinkedHashSet::new)));
versionedGroup.setProcessors(group.getProcessors().stream()
- .map(processor -> mapProcessor(processor, serviceLookup))
+ .map(processor -> mapProcessor(processor, serviceProvider, includedGroupIds, externalControllerServiceReferences))
.collect(Collectors.toCollection(LinkedHashSet::new)));
versionedGroup.setRemoteProcessGroups(group.getRemoteProcessGroups().stream()
@@ -212,7 +226,7 @@ public class NiFiRegistryFlowMapper {
.collect(Collectors.toCollection(LinkedHashSet::new)));
versionedGroup.setProcessGroups(group.getProcessGroups().stream()
- .map(grp -> mapGroup(grp, serviceLookup, registryClient, false, mapDescendantVersionedFlows))
+ .map(grp -> mapGroup(grp, serviceProvider, registryClient, false, mapDescendantVersionedFlows, includedGroupIds, externalControllerServiceReferences))
.collect(Collectors.toCollection(LinkedHashSet::new)));
versionedGroup.setConnections(group.getConnections().stream()
@@ -222,6 +236,10 @@ public class NiFiRegistryFlowMapper {
versionedGroup.setVariables(group.getVariableRegistry().getVariableMap().entrySet().stream()
.collect(Collectors.toMap(entry -> entry.getKey().getName(), Map.Entry::getValue)));
+ if (topLevel) {
+ versionedGroup.setExternalControllerServiceReferences(externalControllerServiceReferences);
+ }
+
return versionedGroup;
}
@@ -314,7 +332,8 @@ public class NiFiRegistryFlowMapper {
return component;
}
- public VersionedControllerService mapControllerService(final ControllerServiceNode controllerService, final ControllerServiceProvider serviceProvider) {
+ public VersionedControllerService mapControllerService(final ControllerServiceNode controllerService, final ControllerServiceProvider serviceProvider, final Set<String> includedGroupIds,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences) {
final VersionedControllerService versionedService = new InstantiatedVersionedControllerService(controllerService.getIdentifier(), controllerService.getProcessGroupIdentifier());
versionedService.setIdentifier(getId(controllerService.getVersionedComponentId(), controllerService.getIdentifier()));
versionedService.setGroupIdentifier(getGroupId(controllerService.getProcessGroupIdentifier()));
@@ -325,7 +344,7 @@ public class NiFiRegistryFlowMapper {
versionedService.setControllerServiceApis(mapControllerServiceApis(controllerService));
versionedService.setProperties(mapProperties(controllerService, serviceProvider));
- versionedService.setPropertyDescriptors(mapPropertyDescriptors(controllerService));
+ versionedService.setPropertyDescriptors(mapPropertyDescriptors(controllerService, serviceProvider, includedGroupIds, externalControllerServiceReferences));
versionedService.setType(controllerService.getCanonicalClassName());
return versionedService;
@@ -357,15 +376,40 @@ public class NiFiRegistryFlowMapper {
return mapped;
}
- private Map<String, VersionedPropertyDescriptor> mapPropertyDescriptors(final ComponentNode component) {
+ private Map<String, VersionedPropertyDescriptor> mapPropertyDescriptors(final ComponentNode component, final ControllerServiceProvider serviceProvider, final Set<String> includedGroupIds,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences) {
final Map<String, VersionedPropertyDescriptor> descriptors = new HashMap<>();
for (final PropertyDescriptor descriptor : component.getProperties().keySet()) {
final VersionedPropertyDescriptor versionedDescriptor = new VersionedPropertyDescriptor();
versionedDescriptor.setName(descriptor.getName());
versionedDescriptor.setDisplayName(descriptor.getDisplayName());
- versionedDescriptor.setIdentifiesControllerService(descriptor.getControllerServiceDefinition() != null);
+
+ final Class<?> referencedServiceType = descriptor.getControllerServiceDefinition();
+ versionedDescriptor.setIdentifiesControllerService(referencedServiceType != null);
+
+ if (referencedServiceType != null) {
+ final String value = component.getProperty(descriptor);
+ if (value != null) {
+ final ControllerServiceNode serviceNode = serviceProvider.getControllerServiceNode(value);
+ if (serviceNode == null) {
+ continue;
+ }
+
+ final String serviceGroupId = serviceNode.getProcessGroupIdentifier();
+ if (!includedGroupIds.contains(serviceGroupId)) {
+ final String serviceId = getId(serviceNode.getVersionedComponentId(), serviceNode.getIdentifier());
+
+ final ExternalControllerServiceReference controllerServiceReference = new ExternalControllerServiceReference();
+ controllerServiceReference.setIdentifier(serviceId);
+ controllerServiceReference.setName(serviceNode.getName());
+ externalControllerServiceReferences.put(serviceId, controllerServiceReference);
+ }
+ }
+ }
+
descriptors.put(descriptor.getName(), versionedDescriptor);
}
+
return descriptors;
}
@@ -452,7 +496,8 @@ public class NiFiRegistryFlowMapper {
return position;
}
- public VersionedProcessor mapProcessor(final ProcessorNode procNode, final ControllerServiceProvider serviceProvider) {
+ public VersionedProcessor mapProcessor(final ProcessorNode procNode, final ControllerServiceProvider serviceProvider, final Set<String> includedGroupIds,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences) {
final VersionedProcessor processor = new InstantiatedVersionedProcessor(procNode.getIdentifier(), procNode.getProcessGroupIdentifier());
processor.setIdentifier(getId(procNode.getVersionedComponentId(), procNode.getIdentifier()));
processor.setGroupIdentifier(getGroupId(procNode.getProcessGroupIdentifier()));
@@ -468,7 +513,7 @@ public class NiFiRegistryFlowMapper {
processor.setPenaltyDuration(procNode.getPenalizationPeriod());
processor.setPosition(mapPosition(procNode.getPosition()));
processor.setProperties(mapProperties(procNode, serviceProvider));
- processor.setPropertyDescriptors(mapPropertyDescriptors(procNode));
+ processor.setPropertyDescriptors(mapPropertyDescriptors(procNode, serviceProvider, includedGroupIds, externalControllerServiceReferences));
processor.setRunDurationMillis(procNode.getRunDuration(TimeUnit.MILLISECONDS));
processor.setSchedulingPeriod(procNode.getSchedulingPeriod());
processor.setSchedulingStrategy(procNode.getSchedulingStrategy().name());
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/BundleUtils.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/BundleUtils.java
index ab89c25..0c26203 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/BundleUtils.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/util/BundleUtils.java
@@ -152,34 +152,34 @@ public final class BundleUtils {
if (versionedGroup.getProcessors() != null) {
versionedGroup.getProcessors().forEach(processor -> {
final BundleCoordinate coordinate = BundleUtils.getCompatibleBundle(extensionManager, processor.getType(), createBundleDto(processor.getBundle()));
-
- final org.apache.nifi.registry.flow.Bundle bundle = new org.apache.nifi.registry.flow.Bundle();
- bundle.setArtifact(coordinate.getId());
- bundle.setGroup(coordinate.getGroup());
- bundle.setVersion(coordinate.getVersion());
- processor.setBundle(bundle);
+ processor.setBundle(createBundle(coordinate));
});
}
if (versionedGroup.getControllerServices() != null) {
versionedGroup.getControllerServices().forEach(controllerService -> {
final BundleCoordinate coordinate = BundleUtils.getCompatibleBundle(extensionManager, controllerService.getType(), createBundleDto(controllerService.getBundle()));
-
- final org.apache.nifi.registry.flow.Bundle bundle = new org.apache.nifi.registry.flow.Bundle();
- bundle.setArtifact(coordinate.getId());
- bundle.setGroup(coordinate.getGroup());
- bundle.setVersion(coordinate.getVersion());
- controllerService.setBundle(bundle);
+ controllerService.setBundle(createBundle(coordinate));
});
}
if (versionedGroup.getProcessGroups() != null) {
- versionedGroup.getProcessGroups().forEach(processGroup -> {
- discoverCompatibleBundles(extensionManager, processGroup);
- });
+ versionedGroup.getProcessGroups().forEach(processGroup -> discoverCompatibleBundles(extensionManager, processGroup));
}
}
+ public static BundleCoordinate discoverCompatibleBundle(final ExtensionManager extensionManager, final String type, final org.apache.nifi.registry.flow.Bundle bundle) {
+ return getCompatibleBundle(extensionManager, type, createBundleDto(bundle));
+ }
+
+ private static org.apache.nifi.registry.flow.Bundle createBundle(final BundleCoordinate coordinate) {
+ final org.apache.nifi.registry.flow.Bundle bundle = new org.apache.nifi.registry.flow.Bundle();
+ bundle.setArtifact(coordinate.getId());
+ bundle.setGroup(coordinate.getGroup());
+ bundle.setVersion(coordinate.getVersion());
+ return bundle;
+ }
+
public static BundleDTO createBundleDto(final org.apache.nifi.registry.flow.Bundle bundle) {
final BundleDTO dto = new BundleDTO();
dto.setArtifact(bundle.getArtifact());
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/versioned/ImportFlowIT.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/versioned/ImportFlowIT.java
index 9320ce5..d1469f3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/versioned/ImportFlowIT.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/integration/versioned/ImportFlowIT.java
@@ -41,6 +41,7 @@ import org.junit.Test;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -136,13 +137,13 @@ public class ImportFlowIT extends FrameworkIntegrationTest {
final Set<VersionedProcessor> versionedProcessors = new HashSet<>();
for (final ProcessorNode processor : processors) {
- final VersionedProcessor versionedProcessor = flowMapper.mapProcessor(processor, getFlowController().getControllerServiceProvider());
+ final VersionedProcessor versionedProcessor = flowMapper.mapProcessor(processor, getFlowController().getControllerServiceProvider(), Collections.emptySet(), new HashMap<>());
versionedProcessors.add(versionedProcessor);
}
final Set<VersionedControllerService> services = new HashSet<>();
for (final ControllerServiceNode serviceNode : controllerServices) {
- final VersionedControllerService service = flowMapper.mapControllerService(serviceNode, getFlowController().getControllerServiceProvider());
+ final VersionedControllerService service = flowMapper.mapControllerService(serviceNode, getFlowController().getControllerServiceProvider(), Collections.emptySet(), new HashMap<>());
services.add(service);
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
index af37577..ff47e5f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
@@ -25,6 +25,7 @@ import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.repository.claim.ContentDirection;
import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
import org.apache.nifi.registry.flow.VersionedFlow;
import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
import org.apache.nifi.registry.flow.VersionedProcessGroup;
@@ -1359,13 +1360,16 @@ public interface NiFiServiceFacade {
* @param registryId the ID of the Flow Registry to persist the snapshot to
* @param flow the flow where the snapshot should be persisted
* @param snapshot the Snapshot to persist
+ * @param externalControllerServiceReferences a mapping of controller service id to ExternalControllerServiceReference for any Controller Service that is referenced in the flow but not included
+ * in the VersionedProcessGroup
* @param comments about the snapshot
* @param expectedVersion the version to save the flow as
* @return the snapshot that represents what was stored in the registry
*
* @throws NiFiCoreException if unable to register the snapshot with the flow registry
*/
- VersionedFlowSnapshot registerVersionedFlowSnapshot(String registryId, VersionedFlow flow, VersionedProcessGroup snapshot, String comments, int expectedVersion);
+ VersionedFlowSnapshot registerVersionedFlowSnapshot(String registryId, VersionedFlow flow, VersionedProcessGroup snapshot,
+ Map<String, ExternalControllerServiceReference> externalControllerServiceReferences, String comments, int expectedVersion);
/**
* Updates the Version Control Information on the Process Group with the given ID
@@ -2157,6 +2161,17 @@ public interface NiFiServiceFacade {
void discoverCompatibleBundles(VersionedProcessGroup versionedGroup);
/**
+ * For any Controller Service that is found in the given Versioned Process Group, if that Controller Service is not itself included in the Versioned Process Groups,
+ * attempts to find an existing Controller Service that matches the definition. If any is found, the component within the Versioned Process Group is updated to point
+ * to the existing service.
+ *
+ * @param versionedFlowSnapshot the flow snapshot
+ * @param parentGroupId the ID of the Process Group from which the Controller Services are inherited
+ * @param user the NiFi user on whose behalf the request is happening; this user is used for validation so that only the Controller Services that the user has READ permissions to are included
+ */
+ void resolveInheritedControllerServices(VersionedFlowSnapshot versionedFlowSnapshot, String parentGroupId, NiFiUser user);
+
+ /**
* @param type the component type
* @param bundleDTO bundle to find the component
* @return the bundle coordinate
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacadeLock.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacadeLock.java
index f7f3b90..67c977f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacadeLock.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacadeLock.java
@@ -16,14 +16,14 @@
*/
package org.apache.nifi.web;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
/**
* Aspect to limit access into the core.
*/
@@ -144,6 +144,12 @@ public class NiFiServiceFacadeLock {
return proceedWithReadLock(proceedingJoinPoint);
}
+ @Around("within(org.apache.nifi.web.NiFiServiceFacade+) && "
+ + "execution(* resolve*(..))")
+ public Object resolveLock(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
+ return proceedWithReadLock(proceedingJoinPoint);
+ }
+
private Object proceedWithReadLock(final ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
final long beforeLock = System.nanoTime();
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
index fc495ef..d9aea86 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
@@ -68,6 +68,7 @@ import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.controller.ComponentNode;
+import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.Counter;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessorNode;
@@ -91,21 +92,27 @@ import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.history.History;
import org.apache.nifi.history.HistoryQuery;
import org.apache.nifi.history.PreviousValue;
+import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.registry.ComponentVariableRegistry;
import org.apache.nifi.registry.authorization.Permissions;
import org.apache.nifi.registry.bucket.Bucket;
import org.apache.nifi.registry.client.NiFiRegistryException;
+import org.apache.nifi.registry.flow.ExternalControllerServiceReference;
import org.apache.nifi.registry.flow.FlowRegistry;
import org.apache.nifi.registry.flow.FlowRegistryClient;
import org.apache.nifi.registry.flow.VersionControlInformation;
import org.apache.nifi.registry.flow.VersionedComponent;
+import org.apache.nifi.registry.flow.VersionedConfigurableComponent;
import org.apache.nifi.registry.flow.VersionedConnection;
+import org.apache.nifi.registry.flow.VersionedControllerService;
import org.apache.nifi.registry.flow.VersionedFlow;
import org.apache.nifi.registry.flow.VersionedFlowCoordinates;
import org.apache.nifi.registry.flow.VersionedFlowSnapshot;
import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata;
import org.apache.nifi.registry.flow.VersionedFlowState;
import org.apache.nifi.registry.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.flow.VersionedProcessor;
+import org.apache.nifi.registry.flow.VersionedPropertyDescriptor;
import org.apache.nifi.registry.flow.diff.ComparableDataFlow;
import org.apache.nifi.registry.flow.diff.ConciseEvolvingDifferenceDescriptor;
import org.apache.nifi.registry.flow.diff.DifferenceType;
@@ -3072,6 +3079,125 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
}
@Override
+ public void resolveInheritedControllerServices(final VersionedFlowSnapshot versionedFlowSnapshot, final String processGroupId, final NiFiUser user) {
+ final VersionedProcessGroup versionedGroup = versionedFlowSnapshot.getFlowContents();
+ resolveInheritedControllerServices(versionedGroup, processGroupId, versionedFlowSnapshot.getExternalControllerServices(), user);
+ }
+
+ private void resolveInheritedControllerServices(final VersionedProcessGroup versionedGroup, final String processGroupId,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences,
+ final NiFiUser user) {
+ final Set<String> availableControllerServiceIds = findAllControllerServiceIds(versionedGroup);
+ final ProcessGroup parentGroup = processGroupDAO.getProcessGroup(processGroupId);
+ final Set<ControllerServiceNode> serviceNodes = parentGroup.getControllerServices(true).stream()
+ .filter(service -> service.isAuthorized(authorizer, RequestAction.READ, user))
+ .collect(Collectors.toSet());
+
+ final ExtensionManager extensionManager = controllerFacade.getExtensionManager();
+ for (final VersionedProcessor processor : versionedGroup.getProcessors()) {
+ final BundleCoordinate compatibleBundle = BundleUtils.discoverCompatibleBundle(extensionManager, processor.getType(), processor.getBundle());
+ final ConfigurableComponent tempComponent = extensionManager.getTempComponent(processor.getType(), compatibleBundle);
+
+ resolveInheritedControllerServices(processor, availableControllerServiceIds, serviceNodes, externalControllerServiceReferences, tempComponent::getPropertyDescriptor);
+ }
+
+ for (final VersionedControllerService service : versionedGroup.getControllerServices()) {
+ final BundleCoordinate compatibleBundle = BundleUtils.discoverCompatibleBundle(extensionManager, service.getType(), service.getBundle());
+ final ConfigurableComponent tempComponent = extensionManager.getTempComponent(service.getType(), compatibleBundle);
+
+ resolveInheritedControllerServices(service, availableControllerServiceIds, serviceNodes, externalControllerServiceReferences, tempComponent::getPropertyDescriptor);
+ }
+
+ for (final VersionedProcessGroup child : versionedGroup.getProcessGroups()) {
+ resolveInheritedControllerServices(child, processGroupId, externalControllerServiceReferences, user);
+ }
+ }
+
+
+ private void resolveInheritedControllerServices(final VersionedConfigurableComponent component, final Set<String> availableControllerServiceIds,
+ final Set<ControllerServiceNode> availableControllerServices,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences,
+ final Function<String, PropertyDescriptor> descriptorLookup) {
+ final Map<String, VersionedPropertyDescriptor> descriptors = component.getPropertyDescriptors();
+ final Map<String, String> properties = component.getProperties();
+
+ resolveInheritedControllerServices(descriptors, properties, availableControllerServiceIds, availableControllerServices, externalControllerServiceReferences, descriptorLookup);
+ }
+
+
+ private void resolveInheritedControllerServices(final Map<String, VersionedPropertyDescriptor> propertyDescriptors, final Map<String, String> componentProperties,
+ final Set<String> availableControllerServiceIds, final Set<ControllerServiceNode> availableControllerServices,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences,
+ final Function<String, PropertyDescriptor> descriptorLookup) {
+
+ for (final Map.Entry<String, String> entry : new HashMap<>(componentProperties).entrySet()) {
+ final String propertyName = entry.getKey();
+ final String propertyValue = entry.getValue();
+
+ final VersionedPropertyDescriptor propertyDescriptor = propertyDescriptors.get(propertyName);
+ if (propertyDescriptor == null) {
+ continue;
+ }
+
+ if (!propertyDescriptor.getIdentifiesControllerService()) {
+ continue;
+ }
+
+ // If the referenced Controller Service is available in this flow, there is nothing to resolve.
+ if (availableControllerServiceIds.contains(propertyValue)) {
+ continue;
+ }
+
+ final ExternalControllerServiceReference externalServiceReference = externalControllerServiceReferences == null ? null : externalControllerServiceReferences.get(propertyValue);
+ if (externalServiceReference == null) {
+ continue;
+ }
+
+ final PropertyDescriptor descriptor = descriptorLookup.apply(propertyName);
+ if (descriptor == null) {
+ continue;
+ }
+
+ final Class<? extends ControllerService> referencedServiceClass = descriptor.getControllerServiceDefinition();
+ if (referencedServiceClass == null) {
+ continue;
+ }
+
+ final String externalControllerServiceName = externalServiceReference.getName();
+ final List<ControllerServiceNode> matchingControllerServices = availableControllerServices.stream()
+ .filter(service -> service.getName().equals(externalControllerServiceName))
+ .filter(service -> referencedServiceClass.isAssignableFrom(service.getProxiedControllerService().getClass()))
+ .collect(Collectors.toList());
+
+ if (matchingControllerServices.size() != 1) {
+ continue;
+ }
+
+ final ControllerServiceNode matchingServiceNode = matchingControllerServices.get(0);
+ final Optional<String> versionedComponentId = matchingServiceNode.getVersionedComponentId();
+ final String resolvedId = versionedComponentId.orElseGet(matchingServiceNode::getIdentifier);
+
+ componentProperties.put(propertyName, resolvedId);
+ }
+ }
+
+ private Set<String> findAllControllerServiceIds(final VersionedProcessGroup group) {
+ final Set<String> ids = new HashSet<>();
+ findAllControllerServiceIds(group, ids);
+ return ids;
+ }
+
+ private void findAllControllerServiceIds(final VersionedProcessGroup group, final Set<String> ids) {
+ for (final VersionedControllerService service : group.getControllerServices()) {
+ ids.add(service.getIdentifier());
+ }
+
+ for (final VersionedProcessGroup childGroup : group.getProcessGroups()) {
+ findAllControllerServiceIds(childGroup, ids);
+ }
+ }
+
+ @Override
public BundleCoordinate getCompatibleBundle(String type, BundleDTO bundleDTO) {
return BundleUtils.getCompatibleBundle(controllerFacade.getExtensionManager(), type, bundleDTO);
}
@@ -3797,7 +3923,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
try {
// add a snapshot to the flow in the registry
- registeredSnapshot = registerVersionedFlowSnapshot(registryId, registeredFlow, versionedProcessGroup, versionedFlowDto.getComments(), snapshotVersion);
+ registeredSnapshot = registerVersionedFlowSnapshot(registryId, registeredFlow, versionedProcessGroup, versionedProcessGroup.getExternalControllerServiceReferences(),
+ versionedFlowDto.getComments(), snapshotVersion);
} catch (final NiFiCoreException e) {
// If the flow has been created, but failed to add a snapshot,
// then we need to capture the created versioned flow information as a partial successful result.
@@ -3977,15 +4104,16 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
}
@Override
- public VersionedFlowSnapshot registerVersionedFlowSnapshot(final String registryId, final VersionedFlow flow,
- final VersionedProcessGroup snapshot, final String comments, final int expectedVersion) {
+ public VersionedFlowSnapshot registerVersionedFlowSnapshot(final String registryId, final VersionedFlow flow, final VersionedProcessGroup snapshot,
+ final Map<String, ExternalControllerServiceReference> externalControllerServiceReferences, final String comments,
+ final int expectedVersion) {
final FlowRegistry registry = flowRegistryClient.getFlowRegistry(registryId);
if (registry == null) {
throw new ResourceNotFoundException("No Flow Registry exists with ID " + registryId);
}
try {
- return registry.registerVersionedFlowSnapshot(flow, snapshot, comments, expectedVersion, NiFiUserUtils.getNiFiUser());
+ return registry.registerVersionedFlowSnapshot(flow, snapshot, externalControllerServiceReferences, comments, expectedVersion, NiFiUserUtils.getNiFiUser());
} catch (final IOException | NiFiRegistryException e) {
throw new NiFiCoreException("Failed to register flow with Flow Registry due to " + e.getMessage(), e);
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
index fe75f7b..9ebd944 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
@@ -1653,7 +1653,7 @@ public class ProcessGroupResource extends ApplicationResource {
@ApiParam(
value = "The process group configuration details.",
required = true
- ) final ProcessGroupEntity requestProcessGroupEntity) throws IOException {
+ ) final ProcessGroupEntity requestProcessGroupEntity) {
if (requestProcessGroupEntity == null || requestProcessGroupEntity.getComponent() == null) {
throw new IllegalArgumentException("Process group details must be specified.");
@@ -1697,21 +1697,14 @@ public class ProcessGroupResource extends ApplicationResource {
if (versionControlInfo != null && requestProcessGroupEntity.getVersionedFlowSnapshot() == null) {
// Step 1: Ensure that user has write permissions to the Process Group. If not, then immediately fail.
// Step 2: Retrieve flow from Flow Registry
- final VersionedFlowSnapshot flowSnapshot = serviceFacade.getVersionedFlowSnapshot(versionControlInfo, true);
- final Bucket bucket = flowSnapshot.getBucket();
- final VersionedFlow flow = flowSnapshot.getFlow();
-
- versionControlInfo.setBucketName(bucket.getName());
- versionControlInfo.setFlowName(flow.getName());
- versionControlInfo.setFlowDescription(flow.getDescription());
-
- versionControlInfo.setRegistryName(serviceFacade.getFlowRegistryName(versionControlInfo.getRegistryId()));
- final VersionedFlowState flowState = flowSnapshot.isLatest() ? VersionedFlowState.UP_TO_DATE : VersionedFlowState.STALE;
- versionControlInfo.setState(flowState.name());
+ final VersionedFlowSnapshot flowSnapshot = getFlowFromRegistry(versionControlInfo);
// Step 3: Resolve Bundle info
serviceFacade.discoverCompatibleBundles(flowSnapshot.getFlowContents());
+ // If there are any Controller Services referenced that are inherited from the parent group, resolve those to point to the appropriate Controller Service, if we are able to.
+ serviceFacade.resolveInheritedControllerServices(flowSnapshot, groupId, NiFiUserUtils.getNiFiUser());
+
// Step 4: Update contents of the ProcessGroupDTO passed in to include the components that need to be added.
requestProcessGroupEntity.setVersionedFlowSnapshot(flowSnapshot);
}
@@ -1791,6 +1784,24 @@ public class ProcessGroupResource extends ApplicationResource {
);
}
+
+ private VersionedFlowSnapshot getFlowFromRegistry(final VersionControlInformationDTO versionControlInfo) {
+ final VersionedFlowSnapshot flowSnapshot = serviceFacade.getVersionedFlowSnapshot(versionControlInfo, true);
+ final Bucket bucket = flowSnapshot.getBucket();
+ final VersionedFlow flow = flowSnapshot.getFlow();
+
+ versionControlInfo.setBucketName(bucket.getName());
+ versionControlInfo.setFlowName(flow.getName());
+ versionControlInfo.setFlowDescription(flow.getDescription());
+
+ versionControlInfo.setRegistryName(serviceFacade.getFlowRegistryName(versionControlInfo.getRegistryId()));
+ final VersionedFlowState flowState = flowSnapshot.isLatest() ? VersionedFlowState.UP_TO_DATE : VersionedFlowState.STALE;
+ versionControlInfo.setState(flowState.name());
+
+ return flowSnapshot;
+ }
+
+
/**
* Retrieves all the processors in this NiFi.
*
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java
index 4fc490c..6d9c42a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/VersionsResource.java
@@ -91,7 +91,6 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
-import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collections;
@@ -1148,6 +1147,9 @@ public class VersionsResource extends ApplicationResource {
// the flow snapshot to contain compatible bundles.
serviceFacade.discoverCompatibleBundles(flowSnapshot.getFlowContents());
+ // If there are any Controller Services referenced that are inherited from the parent group, resolve those to point to the appropriate Controller Service, if we are able to.
+ serviceFacade.resolveInheritedControllerServices(flowSnapshot, groupId, NiFiUserUtils.getNiFiUser());
+
// Step 1: Determine which components will be affected by updating the version
final Set<AffectedComponentEntity> affectedComponents = serviceFacade.getComponentsAffectedByVersionChange(groupId, flowSnapshot);
@@ -1260,7 +1262,7 @@ public class VersionsResource extends ApplicationResource {
@ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.")
})
public Response initiateRevertFlowVersion(@ApiParam("The process group id.") @PathParam("id") final String groupId,
- @ApiParam(value = "The controller service configuration details.", required = true) final VersionControlInformationEntity requestEntity) throws IOException {
+ @ApiParam(value = "The controller service configuration details.", required = true) final VersionControlInformationEntity requestEntity) {
// Verify the request
final RevisionDTO revisionDto = requestEntity.getProcessGroupRevision();
@@ -1309,6 +1311,9 @@ public class VersionsResource extends ApplicationResource {
// the flow snapshot to contain compatible bundles.
serviceFacade.discoverCompatibleBundles(flowSnapshot.getFlowContents());
+ // If there are any Controller Services referenced that are inherited from the parent group, resolve those to point to the appropriate Controller Service, if we are able to.
+ serviceFacade.resolveInheritedControllerServices(flowSnapshot, groupId, NiFiUserUtils.getNiFiUser());
+
// Step 1: Determine which components will be affected by updating the version
final Set<AffectedComponentEntity> affectedComponents = serviceFacade.getComponentsAffectedByVersionChange(groupId, flowSnapshot);