This is an automated email from the ASF dual-hosted git repository.
markap14 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 328aa97bc56 NIFI-15671 - Flow import/export with stateful components
state (#10986)
328aa97bc56 is described below
commit 328aa97bc56867f7f11ff6900fa6572dc782d910
Author: Pierre Villard <[email protected]>
AuthorDate: Mon May 11 21:33:56 2026 +0200
NIFI-15671 - Flow import/export with stateful components state (#10986)
* NIFI-15671 - Flow import/export with stateful components state
Signed-off-by: Pierre Villard <[email protected]>
---
.../http/StandardHttpResponseMapper.java | 2 +
.../ExportProcessGroupEndpointMerger.java | 156 +++++++
.../http/endpoints/ExportedFlowSnapshotEntity.java | 91 ++++
.../StandardVersionedComponentSynchronizer.java | 114 +++++
.../VersionedFlowSynchronizationContext.java | 37 ++
.../apache/nifi/groups/StandardProcessGroup.java | 10 +
.../flow/mapping/VersionedComponentFlowMapper.java | 61 +++
.../nifi/groups/StandardProcessGroupTest.java | 5 +
.../mapping/TestVersionedComponentFlowMapper.java | 106 +++++
.../nifi/controller/ClusterTopologyProvider.java | 40 ++
.../registry/flow/mapping/FlowMappingOptions.java | 64 +++
.../org/apache/nifi/controller/FlowController.java | 40 +-
.../nifi/controller/flow/StandardFlowManager.java | 2 +-
.../org/apache/nifi/web/NiFiServiceFacade.java | 12 +
.../apache/nifi/web/StandardNiFiServiceFacade.java | 87 +++-
.../apache/nifi/web/api/ProcessGroupResource.java | 70 ++-
.../nifi/web/api/TestProcessGroupResource.java | 2 +-
.../stateless/engine/StatelessFlowManager.java | 14 +
.../tests/system/StatefulCountProcessor.java | 70 +++
.../services/org.apache.nifi.processor.Processor | 1 +
.../ClusterFlowDefinitionExportImportStateIT.java | 371 ++++++++++++++++
.../pg/FlowDefinitionExportImportStateIT.java | 484 +++++++++++++++++++++
.../nifi/toolkit/client/ProcessGroupClient.java | 2 +
.../client/impl/JerseyProcessGroupClient.java | 9 +-
24 files changed, 1827 insertions(+), 23 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
index efa017ae5f7..6c2b43a3239 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMapper.java
@@ -44,6 +44,7 @@ import
org.apache.nifi.cluster.coordination.http.endpoints.CountersEndpointMerge
import
org.apache.nifi.cluster.coordination.http.endpoints.CurrentUserEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.DropAllFlowFilesRequestEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.DropRequestEndpointMerger;
+import
org.apache.nifi.cluster.coordination.http.endpoints.ExportProcessGroupEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.FlowAnalysisEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.FlowAnalysisRuleEndpointMerger;
import
org.apache.nifi.cluster.coordination.http.endpoints.FlowAnalysisRuleTypesEndpointMerger;
@@ -161,6 +162,7 @@ public class StandardHttpResponseMapper implements
HttpResponseMapper {
endpointMergers.add(new RemoteProcessGroupsEndpointMerger());
endpointMergers.add(new ProcessGroupEndpointMerger());
endpointMergers.add(new ProcessGroupsEndpointMerger());
+ endpointMergers.add(new ExportProcessGroupEndpointMerger());
endpointMergers.add(new FlowSnippetEndpointMerger());
endpointMergers.add(new PasteEndpointMerger());
endpointMergers.add(new ProvenanceQueryEndpointMerger());
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
new file mode 100644
index 00000000000..22fdd751fba
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportProcessGroupEndpointMerger.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.cluster.coordination.http.endpoints;
+
+import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger;
+import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedConfigurableExtension;
+import org.apache.nifi.flow.VersionedControllerService;
+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 java.net.URI;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * Merges responses for {@code GET /nifi-api/process-groups/{uuid}/download}.
+ * 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.
+ */
+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");
+
+ @Override
+ public boolean canHandle(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) &&
PROCESS_GROUP_DOWNLOAD_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ @Override
+ public NodeResponse merge(final URI uri, final String method, final
Set<NodeResponse> successfulResponses, final Set<NodeResponse>
problematicResponses, final NodeResponse clientResponse) {
+ if (!canHandle(uri, method)) {
+ throw new IllegalArgumentException("Cannot use Endpoint Mapper of
type " + getClass().getSimpleName() + " to map responses for URI " + uri + ",
HTTP Method " + method);
+ }
+
+ final RegisteredFlowSnapshot mergedSnapshot =
clientResponse.getClientResponse().readEntity(RegisteredFlowSnapshot.class);
+ for (final NodeResponse nodeResponse : successfulResponses) {
+ if (nodeResponse == clientResponse) {
+ continue;
+ }
+ final RegisteredFlowSnapshot nodeSnapshot =
nodeResponse.getClientResponse().readEntity(RegisteredFlowSnapshot.class);
+ mergeLocalNodeStates(mergedSnapshot.getFlowContents(),
nodeSnapshot.getFlowContents());
+ }
+
+ return new NodeResponse(clientResponse, new
ExportedFlowSnapshotEntity(mergedSnapshot));
+ }
+
+ /**
+ * Recursively merges localNodeStates from a source process group into a
target process group.
+ * For each stateful component (processor or controller service), the
LOCAL state entries from
+ * the source are added to the target's localNodeStates list. Cluster
state is identical across
+ * nodes and is already present in the target.
+ *
+ * @param target the process group to merge into
+ * @param source the process group to merge from
+ */
+ private void mergeLocalNodeStates(final VersionedProcessGroup target,
final VersionedProcessGroup source) {
+ if (target == null || source == null) {
+ return;
+ }
+
+ if (target.getProcessors() != null && source.getProcessors() != null) {
+ final Map<String, VersionedProcessor> sourceProcessors = new
HashMap<>();
+ for (final VersionedProcessor sp : source.getProcessors()) {
+ sourceProcessors.put(sp.getIdentifier(), sp);
+ }
+ for (final VersionedProcessor tp : target.getProcessors()) {
+ mergeComponentState(tp,
sourceProcessors.get(tp.getIdentifier()));
+ }
+ }
+
+ if (target.getControllerServices() != null &&
source.getControllerServices() != null) {
+ final Map<String, VersionedControllerService> sourceServices = new
HashMap<>();
+ for (final VersionedControllerService ss :
source.getControllerServices()) {
+ sourceServices.put(ss.getIdentifier(), ss);
+ }
+ for (final VersionedControllerService ts :
target.getControllerServices()) {
+ mergeComponentState(ts,
sourceServices.get(ts.getIdentifier()));
+ }
+ }
+
+ if (target.getProcessGroups() != null && source.getProcessGroups() !=
null) {
+ final Map<String, VersionedProcessGroup> sourceGroups = new
HashMap<>();
+ for (final VersionedProcessGroup sg : source.getProcessGroups()) {
+ sourceGroups.put(sg.getIdentifier(), sg);
+ }
+ for (final VersionedProcessGroup tg : target.getProcessGroups()) {
+ mergeLocalNodeStates(tg, sourceGroups.get(tg.getIdentifier()));
+ }
+ }
+ }
+
+ /**
+ * Merges localNodeStates from a source component into a target component.
Each node contributes its
+ * own ordinal entry and the merge fills in any missing slots in the
target's list.
+ *
+ * @param target the target component (already contains state from first
node)
+ * @param source the source component (contains state from another node),
may be {@code null}
+ */
+ private void mergeComponentState(final VersionedConfigurableExtension
target, final VersionedConfigurableExtension source) {
+ if (source == null) {
+ return;
+ }
+
+ final VersionedComponentState sourceState = source.getComponentState();
+ if (sourceState == null || sourceState.getLocalNodeStates() == null) {
+ return;
+ }
+
+ VersionedComponentState targetState = target.getComponentState();
+ if (targetState == null) {
+ targetState = new VersionedComponentState();
+ target.setComponentState(targetState);
+ }
+
+ final List<VersionedNodeState> sourceList =
sourceState.getLocalNodeStates();
+ if (targetState.getLocalNodeStates() == null) {
+ targetState.setLocalNodeStates(new ArrayList<>(sourceList));
+ return;
+ }
+
+ final List<VersionedNodeState> targetList =
targetState.getLocalNodeStates();
+ while (targetList.size() < sourceList.size()) {
+ targetList.add(null);
+ }
+ for (int i = 0; i < sourceList.size(); i++) {
+ if (sourceList.get(i) != null) {
+ targetList.set(i, sourceList.get(i));
+ }
+ }
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportedFlowSnapshotEntity.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportedFlowSnapshotEntity.java
new file mode 100644
index 00000000000..e3496fd2030
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ExportedFlowSnapshotEntity.java
@@ -0,0 +1,91 @@
+/*
+ * 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.xml.bind.annotation.XmlTransient;
+import org.apache.nifi.flow.ExternalControllerServiceReference;
+import org.apache.nifi.flow.ParameterProviderReference;
+import org.apache.nifi.flow.VersionedParameterContext;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.registry.flow.FlowRegistryBucket;
+import org.apache.nifi.registry.flow.RegisteredFlow;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshotMetadata;
+import org.apache.nifi.web.api.entity.Entity;
+
+import java.util.Map;
+
+/**
+ * Internal {@link Entity} wrapper that allows {@link
ExportProcessGroupEndpointMerger} to participate in the
+ * {@code NodeResponse(NodeResponse, Entity)} merging contract while
preserving the on-the-wire JSON format of
+ * {@code GET /process-groups/{id}/download}.
+ *
+ * <p>NiFi's web layer configures Jackson with the JAXB annotation
introspector only, so Jackson-only mechanisms
+ * such as {@code @JsonValue} or {@code @JsonSerialize} are not honored.
Instead, every property of the wrapped
+ * {@link RegisteredFlowSnapshot} is exposed via a delegating getter so JAXB
property discovery produces the same
+ * JSON structure as serializing a {@code RegisteredFlowSnapshot} directly.
+ */
+class ExportedFlowSnapshotEntity extends Entity {
+
+ private final RegisteredFlowSnapshot snapshot;
+
+ ExportedFlowSnapshotEntity(final RegisteredFlowSnapshot snapshot) {
+ this.snapshot = snapshot;
+ }
+
+ @XmlTransient
+ public RegisteredFlowSnapshot getSnapshot() {
+ return snapshot;
+ }
+
+ public RegisteredFlowSnapshotMetadata getSnapshotMetadata() {
+ return snapshot.getSnapshotMetadata();
+ }
+
+ public RegisteredFlow getFlow() {
+ return snapshot.getFlow();
+ }
+
+ public FlowRegistryBucket getBucket() {
+ return snapshot.getBucket();
+ }
+
+ public VersionedProcessGroup getFlowContents() {
+ return snapshot.getFlowContents();
+ }
+
+ public Map<String, ExternalControllerServiceReference>
getExternalControllerServices() {
+ return snapshot.getExternalControllerServices();
+ }
+
+ public Map<String, VersionedParameterContext> getParameterContexts() {
+ return snapshot.getParameterContexts();
+ }
+
+ public String getFlowEncodingVersion() {
+ return snapshot.getFlowEncodingVersion();
+ }
+
+ public Map<String, ParameterProviderReference> getParameterProviders() {
+ return snapshot.getParameterProviders();
+ }
+
+ public boolean isLatest() {
+ return snapshot.isLatest();
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
index 264e2830ae9..e3704ead9ad 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -17,10 +17,15 @@
package org.apache.nifi.flow.synchronization;
+import org.apache.nifi.annotation.behavior.Stateful;
import org.apache.nifi.asset.Asset;
import org.apache.nifi.asset.AssetManager;
import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
@@ -57,6 +62,8 @@ import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.flow.ParameterProviderReference;
import org.apache.nifi.flow.VersionedAsset;
import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedConfigurableExtension;
import org.apache.nifi.flow.VersionedConnection;
import org.apache.nifi.flow.VersionedControllerService;
import org.apache.nifi.flow.VersionedExternalFlow;
@@ -64,6 +71,7 @@ import org.apache.nifi.flow.VersionedFlowAnalysisRule;
import org.apache.nifi.flow.VersionedFlowCoordinates;
import org.apache.nifi.flow.VersionedFunnel;
import org.apache.nifi.flow.VersionedLabel;
+import org.apache.nifi.flow.VersionedNodeState;
import org.apache.nifi.flow.VersionedParameter;
import org.apache.nifi.flow.VersionedParameterContext;
import org.apache.nifi.flow.VersionedPort;
@@ -125,6 +133,8 @@ import org.apache.nifi.util.FlowDifferenceFilters;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.net.URL;
import java.time.Duration;
import java.util.ArrayList;
@@ -327,6 +337,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
createdAndModifiedExtensions.clear();
setSynchronizationOptions(options);
+ validateLocalStateTopology(versionedExternalFlow.getFlowContents());
+
for (final FlowDifference diff : flowComparison.getDifferences()) {
if (!FlowDifferenceFilters.isComponentUpdateRequired(diff,
versionedExternalFlow.getFlowContents(), context.getFlowManager())) {
continue;
@@ -1439,6 +1451,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
updateControllerService(newService, proposed, topLevelGroup);
+ restoreComponentState(newService.getIdentifier(),
proposed.getComponentState(), newService);
+
return newService;
}
@@ -2749,6 +2763,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
procNode.onConfigurationRestored(processContext);
connectableAdditionTracker.addComponent(destination.getIdentifier(),
proposed.getIdentifier(), procNode);
+ restoreComponentState(procNode.getIdentifier(),
proposed.getComponentState(), procNode);
+
return procNode;
}
@@ -4080,6 +4096,104 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
return propertyValues;
}
+ private void validateLocalStateTopology(final VersionedProcessGroup
proposed) {
+ final int connectedNodeCount = context.getConnectedNodeCount();
+ if (connectedNodeCount <= 0) {
+ return;
+ }
+
+ final int maxSourceNodes = findMaxLocalStateNodeCount(proposed);
+ if (maxSourceNodes > connectedNodeCount) {
+ throw new IllegalStateException(
+ "Cannot import flow with component state: the flow
definition contains local state from %d source node(s) but the destination
cluster has only %d connected node(s). "
+ .formatted(maxSourceNodes, connectedNodeCount)
+ + "Import into a cluster with at least %d node(s), or
export without component state.".formatted(maxSourceNodes));
+ }
+ }
+
+ private int findMaxLocalStateNodeCount(final VersionedProcessGroup group) {
+ int max = 0;
+ for (final VersionedConfigurableExtension ext :
getStatefulExtensions(group)) {
+ final VersionedComponentState state = ext.getComponentState();
+ if (state != null && state.getLocalNodeStates() != null) {
+ max = Math.max(max, state.getLocalNodeStates().size());
+ }
+ }
+ if (group.getProcessGroups() != null) {
+ for (final VersionedProcessGroup child : group.getProcessGroups())
{
+ max = Math.max(max, findMaxLocalStateNodeCount(child));
+ }
+ }
+ return max;
+ }
+
+ private List<VersionedConfigurableExtension> getStatefulExtensions(final
VersionedProcessGroup group) {
+ final List<VersionedConfigurableExtension> extensions = new
ArrayList<>();
+ if (group.getProcessors() != null) {
+ extensions.addAll(group.getProcessors());
+ }
+ if (group.getControllerServices() != null) {
+ extensions.addAll(group.getControllerServices());
+ }
+ return extensions;
+ }
+
+ private void restoreComponentState(final String componentId, final
VersionedComponentState componentState, final ComponentNode componentNode) {
+ if (componentState == null) {
+ return;
+ }
+
+ final StateManagerProvider stateManagerProvider =
context.getStateManagerProvider();
+ if (stateManagerProvider == null) {
+ LOG.warn("StateManagerProvider is not available; component state
from the flow snapshot will not be restored for component {}", componentId);
+ return;
+ }
+
+ final ConfigurableComponent component = componentNode.getComponent();
+ if (component == null) {
+ LOG.warn("Component {} is not available; component state from the
flow snapshot will not be restored", componentId);
+ return;
+ }
+
+ final Stateful stateful =
component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ LOG.warn("Component {} ({}) is not annotated with @Stateful;
component state from the flow snapshot will not be restored",
+ componentId, component.getClass().getSimpleName());
+ return;
+ }
+
+ final Set<Scope> supportedScopes = Set.of(stateful.scopes());
+ final StateManager stateManager =
stateManagerProvider.getStateManager(componentId);
+
+ try {
+ if (supportedScopes.contains(Scope.CLUSTER) &&
componentState.getClusterState() != null &&
!componentState.getClusterState().isEmpty()) {
+ stateManager.setState(componentState.getClusterState(),
Scope.CLUSTER);
+ LOG.debug("Restored cluster state for component {}",
componentId);
+ }
+
+ if (supportedScopes.contains(Scope.LOCAL) &&
componentState.getLocalNodeStates() != null &&
!componentState.getLocalNodeStates().isEmpty()) {
+ final int localNodeOrdinal = context.getLocalNodeOrdinal();
+ if (localNodeOrdinal < 0) {
+ LOG.warn("Local node ordinal is not set; local component
state from the flow snapshot will not be restored for component {}",
componentId);
+ return;
+ }
+
+ final List<VersionedNodeState> localNodeStates =
componentState.getLocalNodeStates();
+ final VersionedNodeState nodeState = localNodeOrdinal <
localNodeStates.size() ? localNodeStates.get(localNodeOrdinal) : null;
+ final Map<String, String> localState = nodeState != null ?
nodeState.getState() : null;
+ if (localState != null && !localState.isEmpty()) {
+ stateManager.setState(localState, Scope.LOCAL);
+ LOG.debug("Restored local state for component {} from node
ordinal {}", componentId, localNodeOrdinal);
+ } else {
+ LOG.info("No local component state was captured for node
ordinal {} in the flow snapshot; local state for component {} will not be
restored on this node",
+ localNodeOrdinal, componentId);
+ }
+ }
+ } catch (final IOException e) {
+ throw new UncheckedIOException("Failed to restore state for
component %s".formatted(componentId), e);
+ }
+ }
+
private record CreatedOrModifiedExtension(ComponentNode extension,
Map<String, String> propertyValues) {
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/VersionedFlowSynchronizationContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/VersionedFlowSynchronizationContext.java
index f0b2183b360..0769e304993 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/VersionedFlowSynchronizationContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/VersionedFlowSynchronizationContext.java
@@ -18,6 +18,7 @@
package org.apache.nifi.flow.synchronization;
import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.controller.ComponentNode;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ProcessorNode;
@@ -45,6 +46,9 @@ public class VersionedFlowSynchronizationContext {
private final Function<ProcessorNode, ProcessContext>
processContextFactory;
private final Function<ComponentNode, ConfigurationContext>
configurationContextFactory;
private final AssetManager assetManager;
+ private final StateManagerProvider stateManagerProvider;
+ private final int localNodeOrdinal;
+ private final int connectedNodeCount;
private VersionedFlowSynchronizationContext(final Builder builder) {
this.componentIdGenerator = builder.componentIdGenerator;
@@ -57,6 +61,9 @@ public class VersionedFlowSynchronizationContext {
this.processContextFactory = builder.processContextFactory;
this.configurationContextFactory = builder.configurationContextFactory;
this.assetManager = builder.assetManager;
+ this.stateManagerProvider = builder.stateManagerProvider;
+ this.localNodeOrdinal = builder.localNodeOrdinal;
+ this.connectedNodeCount = builder.connectedNodeCount;
}
public ComponentIdGenerator getComponentIdGenerator() {
@@ -99,6 +106,18 @@ public class VersionedFlowSynchronizationContext {
return assetManager;
}
+ public StateManagerProvider getStateManagerProvider() {
+ return stateManagerProvider;
+ }
+
+ public int getLocalNodeOrdinal() {
+ return localNodeOrdinal;
+ }
+
+ public int getConnectedNodeCount() {
+ return connectedNodeCount;
+ }
+
public static class Builder {
private ComponentIdGenerator componentIdGenerator;
private FlowManager flowManager;
@@ -110,6 +129,9 @@ public class VersionedFlowSynchronizationContext {
private Function<ProcessorNode, ProcessContext> processContextFactory;
private Function<ComponentNode, ConfigurationContext>
configurationContextFactory;
private AssetManager assetManager;
+ private StateManagerProvider stateManagerProvider;
+ private int localNodeOrdinal = -1;
+ private int connectedNodeCount = -1;
public Builder componentIdGenerator(final ComponentIdGenerator
componentIdGenerator) {
this.componentIdGenerator = componentIdGenerator;
@@ -161,6 +183,21 @@ public class VersionedFlowSynchronizationContext {
return this;
}
+ public Builder stateManagerProvider(final StateManagerProvider
stateManagerProvider) {
+ this.stateManagerProvider = stateManagerProvider;
+ return this;
+ }
+
+ public Builder localNodeOrdinal(final int localNodeOrdinal) {
+ this.localNodeOrdinal = localNodeOrdinal;
+ return this;
+ }
+
+ public Builder connectedNodeCount(final int connectedNodeCount) {
+ this.connectedNodeCount = connectedNodeCount;
+ return this;
+ }
+
public VersionedFlowSynchronizationContext build() {
requireNonNull(componentIdGenerator, "Component ID Generator must
be set");
requireNonNull(flowManager, "Flow Manager must be set");
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
index 9f51aa82f0e..755f48f93a7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
@@ -41,6 +41,7 @@ import org.apache.nifi.connectable.Port;
import org.apache.nifi.connectable.Position;
import org.apache.nifi.connectable.Positionable;
import org.apache.nifi.connectable.ProcessGroupFlowFileActivity;
+import org.apache.nifi.controller.ClusterTopologyProvider;
import org.apache.nifi.controller.ComponentNode;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
@@ -195,6 +196,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
private final VersionControlFields versionControlFields = new
VersionControlFields();
private volatile ParameterContext parameterContext;
private final NodeTypeProvider nodeTypeProvider;
+ private final ClusterTopologyProvider clusterTopologyProvider;
private final AssetManager assetManager;
private final StatelessGroupNode statelessGroupNode;
private volatile ExecutionEngine executionEngine =
ExecutionEngine.INHERITED;
@@ -232,6 +234,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
final PropertyEncryptor encryptor, final
ExtensionManager extensionManager,
final StateManagerProvider
stateManagerProvider, final FlowManager flowManager,
final ReloadComponent reloadComponent, final
NodeTypeProvider nodeTypeProvider,
+ final ClusterTopologyProvider
clusterTopologyProvider,
final NiFiProperties nifiProperties, final
StatelessGroupNodeFactory statelessGroupNodeFactory,
final AssetManager assetManager, final String
connectorId) {
@@ -246,6 +249,7 @@ public final class StandardProcessGroup implements
ProcessGroup {
this.flowManager = flowManager;
this.reloadComponent = reloadComponent;
this.nodeTypeProvider = nodeTypeProvider;
+ this.clusterTopologyProvider = clusterTopologyProvider;
this.assetManager = assetManager;
this.connectorId = connectorId;
@@ -4073,6 +4077,9 @@ public final class StandardProcessGroup implements
ProcessGroup {
private VersionedFlowSynchronizationContext
createGroupSynchronizationContext(final ComponentIdGenerator
componentIdGenerator, final ComponentScheduler componentScheduler,
final FlowMappingOptions flowMappingOptions) {
+ final int localNodeOrdinal =
clusterTopologyProvider.getLocalNodeOrdinal();
+ final int connectedNodeCount =
clusterTopologyProvider.getConnectedNodeCount();
+
return new VersionedFlowSynchronizationContext.Builder()
.componentIdGenerator(componentIdGenerator)
.flowManager(flowManager)
@@ -4084,6 +4091,9 @@ public final class StandardProcessGroup implements
ProcessGroup {
.processContextFactory(this::createProcessContext)
.configurationContextFactory(this::createConfigurationContext)
.assetManager(assetManager)
+ .stateManagerProvider(stateManagerProvider)
+ .localNodeOrdinal(localNodeOrdinal)
+ .connectedNodeCount(connectedNodeCount)
.build();
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/registry/flow/mapping/VersionedComponentFlowMapper.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/registry/flow/mapping/VersionedComponentFlowMapper.java
index 1d845265c9b..216e4fd7642 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/registry/flow/mapping/VersionedComponentFlowMapper.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/registry/flow/mapping/VersionedComponentFlowMapper.java
@@ -18,8 +18,10 @@
package org.apache.nifi.registry.flow.mapping;
import org.apache.commons.lang3.ClassUtils;
+import org.apache.nifi.annotation.behavior.Stateful;
import org.apache.nifi.asset.Asset;
import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.components.ConfigurableComponent;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.connector.AssetReference;
import org.apache.nifi.components.connector.ConnectorConfiguration;
@@ -34,6 +36,10 @@ import
org.apache.nifi.components.connector.StringLiteralValue;
import org.apache.nifi.components.listen.ListenPortDefinition;
import org.apache.nifi.components.resource.ResourceCardinality;
import org.apache.nifi.components.resource.ResourceDefinition;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateManagerProvider;
+import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Funnel;
@@ -63,6 +69,7 @@ import org.apache.nifi.flow.ParameterProviderReference;
import org.apache.nifi.flow.PortType;
import org.apache.nifi.flow.Position;
import org.apache.nifi.flow.VersionedAsset;
+import org.apache.nifi.flow.VersionedComponentState;
import org.apache.nifi.flow.VersionedConfigurationStep;
import org.apache.nifi.flow.VersionedConnection;
import org.apache.nifi.flow.VersionedConnector;
@@ -74,6 +81,7 @@ import org.apache.nifi.flow.VersionedFlowRegistryClient;
import org.apache.nifi.flow.VersionedFunnel;
import org.apache.nifi.flow.VersionedLabel;
import org.apache.nifi.flow.VersionedListenPortDefinition;
+import org.apache.nifi.flow.VersionedNodeState;
import org.apache.nifi.flow.VersionedParameter;
import org.apache.nifi.flow.VersionedParameterContext;
import org.apache.nifi.flow.VersionedParameterProvider;
@@ -105,6 +113,8 @@ import
org.apache.nifi.registry.flow.VersionControlInformation;
import org.apache.nifi.remote.PublicPort;
import org.apache.nifi.remote.RemoteGroupPort;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
@@ -518,10 +528,60 @@ public class VersionedComponentFlowMapper {
versionedService.setPropertyDescriptors(mapPropertyDescriptors(controllerService,
serviceProvider, includedGroupIds, externalControllerServiceReferences));
versionedService.setType(controllerService.getCanonicalClassName());
versionedService.setScheduledState(flowMappingOptions.getStateLookup().getState(controllerService));
+
versionedService.setComponentState(mapComponentState(controllerService));
return versionedService;
}
+ private VersionedComponentState mapComponentState(final ComponentNode
componentNode) {
+ if (!flowMappingOptions.isMapComponentState()) {
+ return null;
+ }
+
+ final ConfigurableComponent component = componentNode.getComponent();
+ if (component == null) {
+ return null;
+ }
+
+ final Stateful stateful =
component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ return null;
+ }
+
+ final String componentId = componentNode.getIdentifier();
+ final StateManagerProvider stateManagerProvider =
flowMappingOptions.getStateManagerProvider();
+ final StateManager stateManager =
stateManagerProvider.getStateManager(componentId);
+ final boolean clusterProviderEnabled =
stateManagerProvider.isClusterProviderEnabled();
+ final VersionedComponentState result = new VersionedComponentState();
+ boolean hasState = false;
+
+ try {
+ for (final Scope scope : stateful.scopes()) {
+ if (scope == Scope.CLUSTER && !clusterProviderEnabled) {
+ continue;
+ }
+
+ final StateMap stateMap = stateManager.getState(scope);
+ if (stateMap != null && !stateMap.toMap().isEmpty()) {
+ if (scope == Scope.CLUSTER) {
+ result.setClusterState(stateMap.toMap());
+ hasState = true;
+ } else if (scope == Scope.LOCAL) {
+ final int ordinal =
flowMappingOptions.getLocalNodeOrdinal();
+ final List<VersionedNodeState> localStates = new
ArrayList<>(Collections.nCopies(ordinal + 1, null));
+ localStates.set(ordinal, new
VersionedNodeState(stateMap.toMap()));
+ result.setLocalNodeStates(localStates);
+ hasState = true;
+ }
+ }
+ }
+ } catch (final IOException e) {
+ throw new UncheckedIOException("Failed to retrieve state for
component %s".formatted(componentId), e);
+ }
+
+ return hasState ? result : null;
+ }
+
private Map<String, String> mapProperties(final ComponentNode component,
final ControllerServiceProvider serviceProvider) {
final Map<String, String> mapped = new HashMap<>();
@@ -787,6 +847,7 @@ public class VersionedComponentFlowMapper {
processor.setRetriedRelationships(procNode.getRetriedRelationships());
processor.setBackoffMechanism(procNode.getBackoffMechanism().name());
processor.setMaxBackoffPeriod(procNode.getMaxBackoffPeriod());
+ processor.setComponentState(mapComponentState(procNode));
return processor;
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
index 3a57a120b8e..35051a30f9f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
@@ -21,6 +21,7 @@ import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.controller.ClusterTopologyProvider;
import org.apache.nifi.controller.NodeTypeProvider;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ReloadComponent;
@@ -89,6 +90,9 @@ class StandardProcessGroupTest {
@Mock
private NodeTypeProvider nodeTypeProvider;
+ @Mock
+ private ClusterTopologyProvider clusterTopologyProvider;
+
@Mock
private NiFiProperties properties;
@@ -130,6 +134,7 @@ class StandardProcessGroupTest {
flowManager,
reloadComponent,
nodeTypeProvider,
+ clusterTopologyProvider,
properties,
statelessGroupNodeFactory,
assetManager,
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/registry/flow/mapping/TestVersionedComponentFlowMapper.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/registry/flow/mapping/TestVersionedComponentFlowMapper.java
index cf263592778..9b0677ea35a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/registry/flow/mapping/TestVersionedComponentFlowMapper.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/registry/flow/mapping/TestVersionedComponentFlowMapper.java
@@ -17,10 +17,14 @@
package org.apache.nifi.registry.flow.mapping;
+import org.apache.nifi.annotation.behavior.Stateful;
import org.apache.nifi.attribute.expression.language.VariableImpact;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.connectable.Position;
import org.apache.nifi.controller.BackoffMechanism;
import org.apache.nifi.controller.ControllerService;
@@ -29,17 +33,25 @@ import org.apache.nifi.controller.PropertyConfiguration;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.controller.state.StandardStateMap;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedNodeState;
import org.apache.nifi.flow.VersionedProcessor;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.parameter.ExpressionLanguageAgnosticParameterParser;
import org.apache.nifi.parameter.ParameterReference;
import org.apache.nifi.parameter.ParameterTokenList;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.junit.jupiter.api.Test;
import org.mockito.stubbing.Answer;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -166,6 +178,100 @@ public class TestVersionedComponentFlowMapper {
}
}
+ @Test
+ public void testComponentStateNotMappedWhenDisabled() {
+ final ExtensionManager extensionManager = mock(ExtensionManager.class);
+ final FlowMappingOptions mappingOptions =
FlowMappingOptions.DEFAULT_OPTIONS;
+ final VersionedComponentFlowMapper mapper = new
VersionedComponentFlowMapper(extensionManager, mappingOptions);
+
+ final ProcessorNode procNode = createStatefulProcessorNode();
+
+ final VersionedProcessor versionedProcessor =
mapper.mapProcessor(procNode, mock(ControllerServiceProvider.class),
+ Collections.emptySet(), Collections.emptyMap());
+ assertNotNull(versionedProcessor);
+ assertNull(versionedProcessor.getComponentState());
+ }
+
+ @Test
+ public void
testStandaloneExportSkipsClusterStateForProcessorWithBothScopes() throws
IOException {
+ final Map<String, String> localState = Map.of("wal.position", "0/42");
+ final Map<String, String> clusterState = Map.of("wal.position",
"0/42");
+
+ final ProcessorNode procNode = createStatefulProcessorNode();
+ final FlowMappingOptions mappingOptions =
buildComponentStateMappingOptions(procNode, localState, clusterState, false, 0);
+
+ final VersionedComponentFlowMapper mapper = new
VersionedComponentFlowMapper(mock(ExtensionManager.class), mappingOptions);
+ final VersionedProcessor versionedProcessor =
mapper.mapProcessor(procNode, mock(ControllerServiceProvider.class),
+ Collections.emptySet(), Collections.emptyMap());
+
+ assertNotNull(versionedProcessor);
+ final VersionedComponentState mappedState =
versionedProcessor.getComponentState();
+ assertNotNull(mappedState);
+ assertNull(mappedState.getClusterState());
+
+ final List<VersionedNodeState> localNodeStates =
mappedState.getLocalNodeStates();
+ assertNotNull(localNodeStates);
+ assertEquals(1, localNodeStates.size());
+ assertEquals(localState, localNodeStates.get(0).getState());
+ }
+
+ @Test
+ public void testClusteredExportCapturesBothLocalAndClusterState() throws
IOException {
+ final Map<String, String> localState = Map.of("local.key",
"local-value");
+ final Map<String, String> clusterState = Map.of("cluster.key",
"cluster-value");
+
+ final ProcessorNode procNode = createStatefulProcessorNode();
+ final FlowMappingOptions mappingOptions =
buildComponentStateMappingOptions(procNode, localState, clusterState, true, 1);
+
+ final VersionedComponentFlowMapper mapper = new
VersionedComponentFlowMapper(mock(ExtensionManager.class), mappingOptions);
+ final VersionedProcessor versionedProcessor =
mapper.mapProcessor(procNode, mock(ControllerServiceProvider.class),
+ Collections.emptySet(), Collections.emptyMap());
+
+ assertNotNull(versionedProcessor);
+ final VersionedComponentState mappedState =
versionedProcessor.getComponentState();
+ assertNotNull(mappedState);
+ assertEquals(clusterState, mappedState.getClusterState());
+
+ final List<VersionedNodeState> localNodeStates =
mappedState.getLocalNodeStates();
+ assertNotNull(localNodeStates);
+ assertEquals(2, localNodeStates.size());
+ assertNull(localNodeStates.get(0));
+ assertEquals(localState, localNodeStates.get(1).getState());
+ }
+
+ private FlowMappingOptions buildComponentStateMappingOptions(final
ProcessorNode procNode, final Map<String, String> localState,
+ final
Map<String, String> clusterState, final boolean clusterProviderEnabled,
+ final int
localNodeOrdinal) throws IOException {
+ final StateManager stateManager = mock(StateManager.class);
+ when(stateManager.getState(Scope.LOCAL)).thenReturn(new
StandardStateMap(localState, Optional.empty()));
+ when(stateManager.getState(Scope.CLUSTER)).thenReturn(new
StandardStateMap(clusterState, Optional.empty()));
+
+ final StateManagerProvider stateManagerProvider =
mock(StateManagerProvider.class);
+
when(stateManagerProvider.isClusterProviderEnabled()).thenReturn(clusterProviderEnabled);
+
when(stateManagerProvider.getStateManager(procNode.getIdentifier())).thenReturn(stateManager);
+
+ return new FlowMappingOptions.Builder()
+ .stateLookup(VersionedComponentStateLookup.IDENTITY_LOOKUP)
+ .componentIdLookup(ComponentIdLookup.USE_COMPONENT_ID)
+ .mapComponentState(true)
+ .stateManagerProvider(stateManagerProvider)
+ .localNodeOrdinal(localNodeOrdinal)
+ .build();
+ }
+
+ private ProcessorNode createStatefulProcessorNode() {
+ final ProcessorNode procNode = createProcessorNode(new HashMap<>());
+ when(procNode.getComponent()).thenReturn(new
StatefulLocalAndClusterProcessor());
+ return procNode;
+ }
+
+ @Stateful(scopes = {Scope.LOCAL, Scope.CLUSTER}, description = "Test
processor with both local and cluster state scopes")
+ private static class StatefulLocalAndClusterProcessor extends
AbstractProcessor {
+ @Override
+ public void onTrigger(final ProcessContext context, final
ProcessSession session) throws ProcessException {
+ }
+ }
+
private ProcessorNode createProcessorNode(final Map<String, String>
properties) {
final ExpressionLanguageAgnosticParameterParser parameterParser = new
ExpressionLanguageAgnosticParameterParser();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ClusterTopologyProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ClusterTopologyProvider.java
new file mode 100644
index 00000000000..d85662cb7c9
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ClusterTopologyProvider.java
@@ -0,0 +1,40 @@
+/*
+ * 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.controller;
+
+/**
+ * Framework-internal helper that exposes connected, port-disambiguated
cluster topology information
+ * needed for ordinal-based capture and restoration of component state.
Standalone NiFi instances
+ * report ordinal {@code 0} and a count of {@code 1}.
+ */
+public interface ClusterTopologyProvider {
+
+ /**
+ * Returns the ordinal index of the local node among connected cluster
nodes when sorted
+ * deterministically by API address and API port. Standalone NiFi returns
{@code 0}.
+ *
+ * @return the ordinal index of the local node, or {@code 0} when not
clustered
+ */
+ int getLocalNodeOrdinal();
+
+ /**
+ * Returns the number of currently connected cluster nodes. Standalone
NiFi returns {@code 1}.
+ *
+ * @return the number of connected cluster nodes, or {@code 1} when not
clustered
+ */
+ int getConnectedNodeCount();
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/mapping/FlowMappingOptions.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/mapping/FlowMappingOptions.java
index d6a96eb32e2..eccf808188d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/mapping/FlowMappingOptions.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/registry/flow/mapping/FlowMappingOptions.java
@@ -17,6 +17,8 @@
package org.apache.nifi.registry.flow.mapping;
+import org.apache.nifi.components.state.StateManagerProvider;
+
import static java.util.Objects.requireNonNull;
public class FlowMappingOptions {
@@ -29,6 +31,9 @@ public class FlowMappingOptions {
private final boolean mapControllerServiceReferencesToVersionedId;
private final boolean mapFlowRegistryClientId;
private final boolean mapAssetReferences;
+ private final boolean mapComponentState;
+ private final StateManagerProvider stateManagerProvider;
+ private final int localNodeOrdinal;
private FlowMappingOptions(final Builder builder) {
encryptor = builder.encryptor;
@@ -40,6 +45,9 @@ public class FlowMappingOptions {
mapControllerServiceReferencesToVersionedId =
builder.mapControllerServiceReferencesToVersionedId;
mapFlowRegistryClientId = builder.mapFlowRegistryClientId;
mapAssetReferences = builder.mapAssetReferences;
+ mapComponentState = builder.mapComponentState;
+ stateManagerProvider = builder.stateManagerProvider;
+ localNodeOrdinal = builder.localNodeOrdinal;
}
public SensitiveValueEncryptor getSensitiveValueEncryptor() {
@@ -78,6 +86,18 @@ public class FlowMappingOptions {
return mapAssetReferences;
}
+ public boolean isMapComponentState() {
+ return mapComponentState;
+ }
+
+ public StateManagerProvider getStateManagerProvider() {
+ return stateManagerProvider;
+ }
+
+ public int getLocalNodeOrdinal() {
+ return localNodeOrdinal;
+ }
+
public static class Builder {
private SensitiveValueEncryptor encryptor;
private VersionedComponentStateLookup stateLookup;
@@ -88,6 +108,9 @@ public class FlowMappingOptions {
private boolean mapControllerServiceReferencesToVersionedId = true;
private boolean mapFlowRegistryClientId = false;
private boolean mapAssetReferences = false;
+ private boolean mapComponentState = false;
+ private StateManagerProvider stateManagerProvider;
+ private int localNodeOrdinal = 0;
/**
* Sets the SensitiveValueEncryptor to use for encrypting sensitive
values. This value must be set
@@ -190,6 +213,42 @@ public class FlowMappingOptions {
return this;
}
+ /**
+ * Sets whether or not the component state should be mapped to the
Versioned Component during export.
+ * If <code>true</code>, the {@link
#stateManagerProvider(StateManagerProvider)} must be set.
+ *
+ * @param mapComponentState whether or not component state should be
mapped
+ * @return the builder
+ */
+ public Builder mapComponentState(final boolean mapComponentState) {
+ this.mapComponentState = mapComponentState;
+ return this;
+ }
+
+ /**
+ * Sets the StateManagerProvider to use for retrieving component
state. This value must be set
+ * if {@link #mapComponentState(boolean) mapComponentState} is set to
<code>true</code>.
+ *
+ * @param stateManagerProvider the StateManagerProvider to use
+ * @return the builder
+ */
+ public Builder stateManagerProvider(final StateManagerProvider
stateManagerProvider) {
+ this.stateManagerProvider = stateManagerProvider;
+ return this;
+ }
+
+ /**
+ * Sets the ordinal index of the local node within the cluster. In
standalone mode, this defaults to 0.
+ * Used during export to key local-scoped state entries.
+ *
+ * @param localNodeOrdinal the ordinal index of the local node
+ * @return the builder
+ */
+ public Builder localNodeOrdinal(final int localNodeOrdinal) {
+ this.localNodeOrdinal = localNodeOrdinal;
+ return this;
+ }
+
/**
* Creates a FlowMappingOptions object, or throws an Exception if not
all required configuration has been provided
*
@@ -206,6 +265,10 @@ public class FlowMappingOptions {
requireNonNull(encryptor, "Property Encryptor must be set when
sensitive configuration is to be mapped");
}
+ if (mapComponentState) {
+ requireNonNull(stateManagerProvider, "State Manager Provider
must be set when component state is to be mapped");
+ }
+
return new FlowMappingOptions(this);
}
}
@@ -224,6 +287,7 @@ public class FlowMappingOptions {
.mapControllerServiceReferencesToVersionedId(true)
.mapFlowRegistryClientId(false)
.mapAssetReferences(false)
+ .mapComponentState(false)
.build();
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 8b92b5e411e..aaa8f90109a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -249,6 +249,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
@@ -274,7 +275,7 @@ import javax.net.ssl.SSLContext;
import static java.util.Objects.requireNonNull;
-public class FlowController implements ReportingTaskProvider,
FlowAnalysisRuleProvider, Authorizable, NodeTypeProvider {
+public class FlowController implements ReportingTaskProvider,
FlowAnalysisRuleProvider, Authorizable, NodeTypeProvider,
ClusterTopologyProvider {
private static final String STANDARD_PYTHON_BRIDGE_IMPLEMENTATION_CLASS =
"org.apache.nifi.py4j.StandardPythonBridge";
// default repository implementations
@@ -2962,6 +2963,43 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
}
}
+ @Override
+ public int getLocalNodeOrdinal() {
+ final List<NodeIdentifier> sortedConnectedNodes =
getSortedConnectedNodeIdentifiers();
+ if (sortedConnectedNodes.isEmpty()) {
+ return 0;
+ }
+
+ final NodeIdentifier localNodeId = getNodeId();
+ if (localNodeId == null) {
+ return 0;
+ }
+
+ for (int i = 0; i < sortedConnectedNodes.size(); i++) {
+ if (sortedConnectedNodes.get(i).equals(localNodeId)) {
+ return i;
+ }
+ }
+
+ return 0;
+ }
+
+ @Override
+ public int getConnectedNodeCount() {
+ final List<NodeIdentifier> sortedConnectedNodes =
getSortedConnectedNodeIdentifiers();
+ return sortedConnectedNodes.isEmpty() ? 1 :
sortedConnectedNodes.size();
+ }
+
+ private List<NodeIdentifier> getSortedConnectedNodeIdentifiers() {
+ if (!isClustered() || clusterCoordinator == null) {
+ return Collections.emptyList();
+ }
+
+ return
clusterCoordinator.getNodeIdentifiers(NodeConnectionState.CONNECTED).stream()
+
.sorted(Comparator.comparing(NodeIdentifier::getApiAddress).thenComparingInt(NodeIdentifier::getApiPort))
+ .toList();
+ }
+
@Override
public boolean isConfiguredForClustering() {
return configuredForClustering;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
index 144f3cde8c5..cf74d624310 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/flow/StandardFlowManager.java
@@ -306,7 +306,7 @@ public class StandardFlowManager extends
AbstractFlowManager implements FlowMana
final ProcessGroup group = new
StandardProcessGroup(requireNonNull(id),
flowController.getControllerServiceProvider(), processScheduler,
flowController.getEncryptor(),
flowController.getExtensionManager(),
flowController.getStateManagerProvider(), this,
- flowController.getReloadComponent(), flowController,
nifiProperties, statelessGroupNodeFactory,
+ flowController.getReloadComponent(), flowController,
flowController, nifiProperties, statelessGroupNodeFactory,
flowController.getAssetManager(), connectorId);
onProcessGroupAdded(group);
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 10f5c9c585c..5a2d96ee2a6 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
@@ -1845,6 +1845,18 @@ public interface NiFiServiceFacade {
*/
RegisteredFlowSnapshot
getCurrentFlowSnapshotByGroupIdWithReferencedControllerServices(String
processGroupId);
+ /**
+ * Get the current state of the Process Group with the given ID, converted
to a Versioned Flow Snapshot for download.
+ * Optionally includes referenced controller services from parent groups
and component state.
+ *
+ * @param processGroupId the ID of the Process Group
+ * @param includeReferencedServices whether to include referenced
controller services from parent groups
+ * @param includeComponentState whether to include component state in the
export. When true, all processors must be stopped
+ * and all controller services must be
disabled.
+ * @return the current Process Group converted to a Versioned Flow
Snapshot for download
+ */
+ RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(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 1b03128abe4..784710c3a34 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
@@ -85,12 +85,14 @@ import
org.apache.nifi.components.connector.ConnectorUpdateContext;
import org.apache.nifi.components.connector.Secret;
import org.apache.nifi.components.connector.secrets.AuthorizableSecret;
import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.components.validation.ValidationState;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.connectable.Port;
+import org.apache.nifi.controller.ClusterTopologyProvider;
import org.apache.nifi.controller.ComponentNode;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.Counter;
@@ -517,6 +519,8 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
private FlowRegistryDAO flowRegistryDAO;
private ParameterContextDAO parameterContextDAO;
private ClusterCoordinator clusterCoordinator;
+ private ClusterTopologyProvider clusterTopologyProvider;
+ private StateManagerProvider stateManagerProvider;
private HeartbeatMonitor heartbeatMonitor;
private LeaderElectionManager leaderElectionManager;
@@ -5927,6 +5931,51 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
return getCurrentFlowSnapshotByGroupId(processGroupId, true);
}
+ @Override
+ public RegisteredFlowSnapshot getCurrentFlowSnapshotByGroupId(final String
processGroupId, final boolean includeReferencedServices, final boolean
includeComponentState) {
+ if (!includeComponentState) {
+ return getCurrentFlowSnapshotByGroupId(processGroupId,
includeReferencedServices);
+ }
+
+ 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)
+ .toList();
+ if (!runningProcessors.isEmpty()) {
+ throw new IllegalStateException("Cannot export component state
because %d processor(s) are not stopped: %s".formatted(
+ runningProcessors.size(),
+ runningProcessors.stream().map(p -> "%s
[%s]".formatted(p.getName(),
p.getIdentifier())).limit(5).collect(Collectors.joining(", "))));
+ }
+
+ final List<ControllerServiceNode> enabledServices =
processGroup.findAllControllerServices().stream()
+ .filter(s -> s.getState() != ControllerServiceState.DISABLED)
+ .toList();
+ if (!enabledServices.isEmpty()) {
+ throw new IllegalStateException("Cannot export component state
because %d controller service(s) are not disabled: %s".formatted(
+ enabledServices.size(),
+ enabledServices.stream().map(s -> "%s
[%s]".formatted(s.getName(),
s.getIdentifier())).limit(5).collect(Collectors.joining(", "))));
+ }
+
+ final FlowMappingOptions mappingOptions = new
FlowMappingOptions.Builder()
+ .sensitiveValueEncryptor(null)
+ .stateLookup(VersionedComponentStateLookup.ENABLED_OR_DISABLED)
+ .componentIdLookup(ComponentIdLookup.VERSIONED_OR_GENERATE)
+ .mapPropertyDescriptors(true)
+ .mapSensitiveConfiguration(false)
+ .mapInstanceIdentifiers(false)
+ .mapControllerServiceReferencesToVersionedId(true)
+ .mapFlowRegistryClientId(false)
+ .mapAssetReferences(false)
+ .mapComponentState(includeComponentState)
+ .stateManagerProvider(stateManagerProvider)
+
.localNodeOrdinal(clusterTopologyProvider.getLocalNodeOrdinal())
+ .build();
+
+ return buildFlowSnapshot(processGroup, processGroupId,
includeReferencedServices, mappingOptions);
+ }
+
private Set<String> getAllSubGroups(ProcessGroup processGroup) {
final Set<String> result = processGroup.findAllProcessGroups().stream()
.map(ProcessGroup::getIdentifier)
@@ -5937,22 +5986,27 @@ 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);
+ }
+ private RegisteredFlowSnapshot buildFlowSnapshot(final ProcessGroup
processGroup, final String processGroupId,
+ final boolean
includeReferencedControllerServices, final FlowMappingOptions
customMappingOptions) {
// Create a complete (include descendant flows) VersionedProcessGroup
snapshot of the flow as it is
// currently without any registry related fields populated, even if
the flow is currently versioned.
- final VersionedComponentFlowMapper mapper =
makeNiFiRegistryFlowMapper(controllerFacade.getExtensionManager());
+ final VersionedComponentFlowMapper mapper = customMappingOptions !=
null
+ ?
makeNiFiRegistryFlowMapper(controllerFacade.getExtensionManager(),
customMappingOptions)
+ :
makeNiFiRegistryFlowMapper(controllerFacade.getExtensionManager());
+
final InstantiatedVersionedProcessGroup nonVersionedProcessGroup =
mapper.mapNonVersionedProcessGroup(processGroup,
controllerFacade.getControllerServiceProvider());
final Map<String, ParameterProviderReference>
parameterProviderReferences = new HashMap<>();
-
- // Create a complete (include descendant flows) map of parameter
contexts
final Map<String, VersionedParameterContext> parameterContexts =
mapper.mapParameterContexts(processGroup, true, parameterProviderReferences);
final Map<String, ExternalControllerServiceReference>
externalControllerServiceReferences =
Optional.ofNullable(nonVersionedProcessGroup.getExternalControllerServiceReferences()).orElse(Collections.emptyMap());
final Set<VersionedControllerService> controllerServices = new
HashSet<>(nonVersionedProcessGroup.getControllerServices());
- final RegisteredFlowSnapshot nonVersionedFlowSnapshot = new
RegisteredFlowSnapshot();
+ final RegisteredFlowSnapshot flowSnapshot = new
RegisteredFlowSnapshot();
ProcessGroup parentGroup = processGroup.getParent();
@@ -5968,24 +6022,25 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
if
(externalControllerServiceReferences.keySet().contains(versionedControllerService.getIdentifier()))
{
versionedControllerService.setGroupIdentifier(processGroupId);
+ versionedControllerService.setComponentState(null);
externalServices.add(versionedControllerService);
}
}
} while ((parentGroup = parentGroup.getParent()) != null);
controllerServices.addAll(externalServices);
- nonVersionedFlowSnapshot.setExternalControllerServices(new
HashMap<>());
+ flowSnapshot.setExternalControllerServices(new HashMap<>());
} else {
-
nonVersionedFlowSnapshot.setExternalControllerServices(externalControllerServiceReferences);
+
flowSnapshot.setExternalControllerServices(externalControllerServiceReferences);
}
nonVersionedProcessGroup.setControllerServices(controllerServices);
- nonVersionedFlowSnapshot.setFlowContents(nonVersionedProcessGroup);
-
nonVersionedFlowSnapshot.setParameterProviders(parameterProviderReferences);
- nonVersionedFlowSnapshot.setParameterContexts(parameterContexts);
-
nonVersionedFlowSnapshot.setFlowEncodingVersion(FlowRegistryUtil.FLOW_ENCODING_VERSION);
+ flowSnapshot.setFlowContents(nonVersionedProcessGroup);
+ flowSnapshot.setParameterProviders(parameterProviderReferences);
+ flowSnapshot.setParameterContexts(parameterContexts);
+
flowSnapshot.setFlowEncodingVersion(FlowRegistryUtil.FLOW_ENCODING_VERSION);
- return nonVersionedFlowSnapshot;
+ return flowSnapshot;
}
@Override
@@ -7929,6 +7984,16 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
this.clusterCoordinator = coordinator;
}
+ @Autowired
+ public void setClusterTopologyProvider(final ClusterTopologyProvider
clusterTopologyProvider) {
+ this.clusterTopologyProvider = clusterTopologyProvider;
+ }
+
+ @Autowired
+ public void setStateManagerProvider(final StateManagerProvider
stateManagerProvider) {
+ this.stateManagerProvider = stateManagerProvider;
+ }
+
@Autowired(required = false)
public void setHeartbeatMonitor(final HeartbeatMonitor heartbeatMonitor) {
this.heartbeatMonitor = heartbeatMonitor;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
index 05807925eef..c42ff3273a0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
@@ -62,9 +62,11 @@ import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.flow.ConnectableComponent;
import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.flow.VersionedComponent;
+import org.apache.nifi.flow.VersionedControllerService;
import org.apache.nifi.flow.VersionedFlowCoordinates;
import org.apache.nifi.flow.VersionedParameterContext;
import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
import org.apache.nifi.flow.VersionedPropertyDescriptor;
import org.apache.nifi.groups.VersionedComponentAdditions;
import org.apache.nifi.parameter.ParameterContext;
@@ -290,7 +292,8 @@ public class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
@ApiResponse(responseCode = "409", description = "The
request was valid but NiFi was not in the appropriate state to process it.")
},
security = {
- @SecurityRequirement(name = "Read -
/process-groups/{uuid}")
+ @SecurityRequirement(name = "Read -
/process-groups/{uuid}"),
+ @SecurityRequirement(name = "Write -
/process-groups/{uuid} - Only required when includeComponentState is true")
}
)
public Response exportProcessGroup(
@@ -301,19 +304,35 @@ public class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
@PathParam("id") final String groupId,
@Parameter(description = "If referenced services from outside the
target group should be included")
@QueryParam("includeReferencedServices")
- @DefaultValue("false") boolean includeReferencedServices) {
- // authorize access
+ @DefaultValue("false") final boolean includeReferencedServices,
+ @Parameter(description = "If component state should be included in
the exported flow definition. "
+ + "Requires all processors to be stopped and all
controller services to be disabled.")
+ @QueryParam("includeComponentState")
+ @DefaultValue("false") final boolean includeComponentState) {
+
+ // When exporting with component state in a cluster, replicate to all
nodes so that each contributes its
+ // LOCAL state. The framework's ExportProcessGroupEndpointMerger
combines the per-node localNodeStates.
+ if (includeComponentState && isReplicateRequest()) {
+ return replicate(HttpMethod.GET);
+ }
+
+ // authorize access — exporting with component state requires WRITE
(state access requires write permission)
+ final RequestAction requiredAction = includeComponentState ?
RequestAction.WRITE : RequestAction.READ;
serviceFacade.authorizeAccess(lookup -> {
- // ensure access to process groups (nested), encapsulated
controller services and referenced parameter contexts
final ProcessGroupAuthorizable groupAuthorizable =
lookup.getProcessGroup(groupId);
- authorizeProcessGroup(groupAuthorizable, authorizer, lookup,
RequestAction.READ, true,
+ authorizeProcessGroup(groupAuthorizable, authorizer, lookup,
requiredAction, true,
false, false, false, true);
});
// get the versioned flow
- final RegisteredFlowSnapshot currentVersionedFlowSnapshot =
includeReferencedServices
- ?
serviceFacade.getCurrentFlowSnapshotByGroupIdWithReferencedControllerServices(groupId)
- : serviceFacade.getCurrentFlowSnapshotByGroupId(groupId);
+ final RegisteredFlowSnapshot currentVersionedFlowSnapshot;
+ if (includeComponentState) {
+ currentVersionedFlowSnapshot =
serviceFacade.getCurrentFlowSnapshotByGroupId(groupId,
includeReferencedServices, true);
+ } else if (includeReferencedServices) {
+ currentVersionedFlowSnapshot =
serviceFacade.getCurrentFlowSnapshotByGroupIdWithReferencedControllerServices(groupId);
+ } else {
+ currentVersionedFlowSnapshot =
serviceFacade.getCurrentFlowSnapshotByGroupId(groupId);
+ }
// determine the name of the attachment - possible issues with spaces
in file names
final VersionedProcessGroup currentVersionedProcessGroup =
currentVersionedFlowSnapshot.getFlowContents();
@@ -2582,6 +2601,11 @@ public class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
throw new IllegalArgumentException("Versioned Flow Snapshot must
be supplied");
}
+ if (containsComponentState(versionedFlowSnapshot.getFlowContents())) {
+ throw new IllegalArgumentException("Cannot replace an existing
Process Group with a flow definition that contains component state. "
+ + "Component state can only be restored when uploading a
flow definition as a new Process Group.");
+ }
+
// remove any registry-specific versioning content which could be
present if the flow was exported from registry
versionedFlowSnapshot.setFlow(null);
versionedFlowSnapshot.setBucket(null);
@@ -2606,6 +2630,31 @@ public class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
}
}
+ private boolean containsComponentState(final VersionedProcessGroup group) {
+ if (group.getProcessors() != null) {
+ for (final VersionedProcessor processor : group.getProcessors()) {
+ if (processor.getComponentState() != null) {
+ return true;
+ }
+ }
+ }
+ if (group.getControllerServices() != null) {
+ for (final VersionedControllerService service :
group.getControllerServices()) {
+ if (service.getComponentState() != null) {
+ return true;
+ }
+ }
+ }
+ if (group.getProcessGroups() != null) {
+ for (final VersionedProcessGroup child : group.getProcessGroups())
{
+ if (containsComponentState(child)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
/**
* Uploads the specified versioned flow definition and adds it to a new
process group.
*
@@ -3245,6 +3294,11 @@ public class ProcessGroupResource extends
FlowUpdateResource<ProcessGroupImportE
throw new IllegalArgumentException("Versioned Flow Snapshot must
be supplied.");
}
+ if (containsComponentState(requestFlowSnapshot.getFlowContents())) {
+ throw new IllegalArgumentException("Cannot replace an existing
Process Group with a flow definition that contains component state. "
+ + "Component state can only be restored when uploading a
flow definition as a new Process Group.");
+ }
+
// Perform the request
if (isReplicateRequest()) {
return replicate(HttpMethod.PUT, importEntity);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestProcessGroupResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestProcessGroupResource.java
index f34f0704afe..7fb43da69f8 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestProcessGroupResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestProcessGroupResource.java
@@ -60,7 +60,7 @@ public class TestProcessGroupResource {
when(versionedFlowSnapshot.getFlowContents()).thenReturn(versionedProcessGroup);
when(versionedProcessGroup.getName()).thenReturn("flowname");
- try (Response response =
processGroupResource.exportProcessGroup(groupId, false)) {
+ try (Response response =
processGroupResource.exportProcessGroup(groupId, false, false)) {
assertEquals(200, response.getStatus());
assertEquals(versionedFlowSnapshot, response.getEntity());
}
diff --git
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
index a637745bb61..cc5c78688e0 100644
---
a/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
+++
b/nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/engine/StatelessFlowManager.java
@@ -30,6 +30,7 @@ import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.connectable.LocalPort;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.connectable.StandardConnection;
+import org.apache.nifi.controller.ClusterTopologyProvider;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.FlowAnalysisRuleNode;
@@ -93,6 +94,18 @@ import static java.util.Objects.requireNonNull;
public class StatelessFlowManager extends AbstractFlowManager implements
FlowManager {
private static final Logger logger =
LoggerFactory.getLogger(StatelessFlowManager.class);
+ private static final ClusterTopologyProvider
STANDALONE_CLUSTER_TOPOLOGY_PROVIDER = new ClusterTopologyProvider() {
+ @Override
+ public int getLocalNodeOrdinal() {
+ return 0;
+ }
+
+ @Override
+ public int getConnectedNodeCount() {
+ return 1;
+ }
+ };
+
private final StatelessEngine statelessEngine;
private final SSLContext sslContext;
private final BulletinRepository bulletinRepository;
@@ -235,6 +248,7 @@ public class StatelessFlowManager extends
AbstractFlowManager implements FlowMan
this,
statelessEngine.getReloadComponent(),
new StatelessNodeTypeProvider(),
+ STANDALONE_CLUSTER_TOPOLOGY_PROVIDER,
null,
group -> null,
statelessEngine.getAssetManager(),
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/StatefulCountProcessor.java
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/StatefulCountProcessor.java
new file mode 100644
index 00000000000..85d13228e0c
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/java/org/apache/nifi/processors/tests/system/StatefulCountProcessor.java
@@ -0,0 +1,70 @@
+/*
+ * 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.processors.tests.system;
+
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.configuration.DefaultSchedule;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
+
+@DefaultSchedule(period = "100 ms")
+@Stateful(scopes = {Scope.CLUSTER, Scope.LOCAL}, description = "Stores a
counter in both cluster and local state")
+public class StatefulCountProcessor extends AbstractProcessor {
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .build();
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return Set.of(REL_SUCCESS);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ flowFile = session.create();
+ }
+
+ try {
+ incrementState(session, Scope.CLUSTER);
+ incrementState(session, Scope.LOCAL);
+ } catch (final IOException e) {
+ throw new ProcessException(e);
+ }
+
+ session.transfer(flowFile, REL_SUCCESS);
+ }
+
+ private void incrementState(final ProcessSession session, final Scope
scope) throws IOException {
+ final StateMap stateMap = session.getState(scope);
+ final String countValue = stateMap.toMap().get("count");
+ final int count = countValue == null ? 0 :
Integer.parseInt(countValue);
+ session.setState(Map.of("count", String.valueOf(count + 1)), scope);
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 2895fadb5c0..80655da6f36 100644
---
a/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-system-tests/nifi-system-test-extensions-bundle/nifi-system-test-extensions/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -50,6 +50,7 @@ org.apache.nifi.processors.tests.system.SetState
org.apache.nifi.processors.tests.system.Sleep
org.apache.nifi.processors.tests.system.SplitByLine
org.apache.nifi.processors.tests.system.SplitTextByLine
+org.apache.nifi.processors.tests.system.StatefulCountProcessor
org.apache.nifi.processors.tests.system.TerminateFlowFile
org.apache.nifi.processors.tests.system.TransferBatch
org.apache.nifi.processors.tests.system.ThrowExceptionInFlowFileFilter
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java
new file mode 100644
index 00000000000..8eb586b77ff
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java
@@ -0,0 +1,371 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.tests.system.pg;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.flow.VersionedComponentState;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
+import org.apache.nifi.tests.system.NiFiInstanceFactory;
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+import org.apache.nifi.web.api.dto.StateEntryDTO;
+import org.apache.nifi.web.api.entity.ComponentStateEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupImportEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupReplaceRequestEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ClusterFlowDefinitionExportImportStateIT extends NiFiSystemIT {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ return createTwoNodeInstanceFactory();
+ }
+
+ @Test
+ public void testClusterExportCapturesClusterState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("GenerateFlowFile", pg.getId());
+ getClientUtil().updateProcessorProperties(stateful,
Collections.singletonMap("State Scope", "CLUSTER"));
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st13-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "GenerateFlowFile");
+ assertNotNull(proc);
+ assertNotNull(proc.getComponentState());
+ assertNotNull(proc.getComponentState().getClusterState());
+ assertNotNull(proc.getComponentState().getClusterState().get("count"));
+ }
+
+ @Test
+ public void testClusterExportCapturesLocalStateFromBothNodes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("GenerateFlowFile", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForLocalStatePopulatedOnAllNodes(stateful.getId(), 2);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st14-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "GenerateFlowFile");
+ assertNotNull(proc);
+ assertNotNull(proc.getComponentState());
+ assertNotNull(proc.getComponentState().getLocalNodeStates());
+ assertEquals(2, proc.getComponentState().getLocalNodeStates().size(),
+ "Should have local state from both nodes");
+ assertNotNull(proc.getComponentState().getLocalNodeStates().get(0));
+ assertNotNull(proc.getComponentState().getLocalNodeStates().get(1));
+ }
+
+ @Test
+ public void testClusterExportCapturesBothScopes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ waitForLocalStatePopulatedOnAllNodes(stateful.getId(), 2);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st15-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "StatefulCountProcessor");
+ assertNotNull(proc);
+
+ final VersionedComponentState state = proc.getComponentState();
+ assertNotNull(state);
+ assertNotNull(state.getClusterState(), "Cluster state should be
present");
+ assertNotNull(state.getClusterState().get("count"), "Cluster state
should contain count");
+ assertNotNull(state.getLocalNodeStates(), "Local node states should be
present");
+ assertEquals(2, state.getLocalNodeStates().size());
+
assertNotNull(state.getLocalNodeStates().get(0).getState().get("count"));
+
assertNotNull(state.getLocalNodeStates().get(1).getState().get("count"));
+ }
+
+ @Test
+ public void testClusterRoundTripSameTopologyBothScopes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ waitForLocalStatePopulatedOnAllNodes(stateful.getId(), 2);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final Map<String, String> originalClusterState =
getProcessorState(stateful.getId(), Scope.CLUSTER);
+ final Map<String, Map<String, String>> originalLocalStatesByNode =
getProcessorLocalStatesByNode(stateful.getId());
+ assertEquals(2, originalLocalStatesByNode.size(),
+ "Local state should be populated on both cluster nodes prior
to export");
+
+ final File exportFile = new File("target/st16-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ emptyQueuesAndDeleteProcessGroup(pg);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedGroup", 0.0, 0.0);
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(uploaded.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+
+ waitForLocalStatePopulatedOnAllNodes(importedProcessor.getId(), 2);
+
+ final Map<String, String> importedClusterState =
getProcessorState(importedProcessor.getId(), Scope.CLUSTER);
+ assertEquals(originalClusterState.get("count"),
importedClusterState.get("count"),
+ "Cluster state should be restored after round-trip");
+
+ final Map<String, Map<String, String>> importedLocalStatesByNode =
getProcessorLocalStatesByNode(importedProcessor.getId());
+ assertEquals(originalLocalStatesByNode, importedLocalStatesByNode,
+ "Local state should be restored to the same node it was
exported from after round-trip");
+ }
+
+ @Test
+ public void testClusterExportRunningProcessorReturns409() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+
+ try {
+ final File exportFile = new File("target/st20-export.json");
+
getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(), false,
true, exportFile);
+ throw new AssertionError("Expected export to fail when processors
are running");
+ } catch (final NiFiClientException e) {
+ assertNotNull(e.getMessage());
+ } finally {
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+ }
+ }
+
+ @Test
+ public void testClusterReplaceRejectsFlowDefinitionWithComponentState()
throws NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st21-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final NiFiClientException exception =
assertThrows(NiFiClientException.class, () -> replaceProcessGroup(pg,
snapshot));
+ assertTrue(exception.getMessage().contains("component state"),
+ "Expected rejection message about component state but got: " +
exception.getMessage());
+ }
+
+ @Test
+ public void testClusterImportWithoutComponentStateHasNoState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.CLUSTER);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st22-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, false, exportFile);
+
+ emptyQueuesAndDeleteProcessGroup(pg);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedGroup", 0.0, 0.0);
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(uploaded.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+
+ final Map<String, String> importedClusterState =
getProcessorState(importedProcessor.getId(), Scope.CLUSTER);
+ assertTrue(importedClusterState.isEmpty(), "State should be empty when
imported without componentState");
+ }
+
+ private void waitForStatePopulated(final String processorId, final Scope
scope) throws InterruptedException {
+ waitFor(() -> {
+ try {
+ final Map<String, String> state =
getProcessorState(processorId, scope);
+ return state.get("count") != null;
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+ }
+
+ private void waitForLocalStatePopulatedOnAllNodes(final String
processorId, final int expectedNodeCount) throws InterruptedException {
+ waitFor(() -> {
+ try {
+ final Map<String, Map<String, String>> byNode =
getProcessorLocalStatesByNode(processorId);
+ if (byNode.size() != expectedNodeCount) {
+ return false;
+ }
+ for (final Map<String, String> nodeState : byNode.values()) {
+ if (nodeState.get("count") == null) {
+ return false;
+ }
+ }
+ return true;
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+ }
+
+ private Map<String, Map<String, String>>
getProcessorLocalStatesByNode(final String processorId) throws
NiFiClientException, IOException {
+ final ComponentStateEntity stateEntity =
getNifiClient().getProcessorClient().getProcessorState(processorId);
+ final ComponentStateDTO componentState =
stateEntity.getComponentState();
+ final Map<String, Map<String, String>> byNode = new HashMap<>();
+ if (componentState != null && componentState.getLocalState() != null
&& componentState.getLocalState().getState() != null) {
+ for (final StateEntryDTO entry :
componentState.getLocalState().getState()) {
+ final String nodeId = entry.getClusterNodeId();
+ byNode.computeIfAbsent(nodeId, id -> new
HashMap<>()).put(entry.getKey(), entry.getValue());
+ }
+ }
+ return byNode;
+ }
+
+ private Map<String, String> getProcessorState(final String processorId,
final Scope scope) throws NiFiClientException, IOException {
+ final ComponentStateEntity stateEntity =
getNifiClient().getProcessorClient().getProcessorState(processorId);
+ final ComponentStateDTO componentState =
stateEntity.getComponentState();
+ final Map<String, String> result = new HashMap<>();
+
+ switch (scope) {
+ case LOCAL:
+ if (componentState != null && componentState.getLocalState()
!= null && componentState.getLocalState().getState() != null) {
+ for (final StateEntryDTO entry :
componentState.getLocalState().getState()) {
+ result.put(entry.getKey(), entry.getValue());
+ }
+ }
+ break;
+ case CLUSTER:
+ if (componentState != null && componentState.getClusterState()
!= null && componentState.getClusterState().getState() != null) {
+ for (final StateEntryDTO entry :
componentState.getClusterState().getState()) {
+ result.put(entry.getKey(), entry.getValue());
+ }
+ }
+ break;
+ }
+ return result;
+ }
+
+ private VersionedProcessor findProcessorByType(final VersionedProcessGroup
group, final String typeSuffix) {
+ if (group.getProcessors() != null) {
+ for (final VersionedProcessor vp : group.getProcessors()) {
+ if (vp.getType() != null && vp.getType().endsWith(typeSuffix))
{
+ return vp;
+ }
+ }
+ }
+ return null;
+ }
+
+ private ProcessorEntity findProcessorByTypeInGroup(final String groupId,
final String typeSuffix) throws NiFiClientException, IOException {
+ return getNifiClient().getFlowClient().getProcessGroup(groupId)
+ .getProcessGroupFlow().getFlow().getProcessors().stream()
+ .filter(pe -> pe.getComponent().getType().endsWith(typeSuffix))
+ .findFirst().orElse(null);
+ }
+
+ private void emptyQueuesAndDeleteProcessGroup(final ProcessGroupEntity pg)
throws NiFiClientException, IOException, InterruptedException {
+ getNifiClient().getProcessGroupClient().emptyQueues(pg.getId());
+ waitFor(() -> {
+ try {
+ return
getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId())
+
.getStatus().getAggregateSnapshot().getQueuedCount().equals("0");
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+ final ProcessGroupEntity refreshed =
getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId());
+ refreshed.setDisconnectedNodeAcknowledged(true);
+ getNifiClient().getProcessGroupClient().deleteProcessGroup(refreshed);
+ }
+
+ private void replaceProcessGroup(final ProcessGroupEntity pg, final
RegisteredFlowSnapshot snapshot) throws NiFiClientException, IOException,
InterruptedException {
+ final ProcessGroupImportEntity importEntity = new
ProcessGroupImportEntity();
+ importEntity.setVersionedFlowSnapshot(snapshot);
+
importEntity.setProcessGroupRevision(getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId()).getRevision());
+
+ final ProcessGroupReplaceRequestEntity replaceRequest =
+
getNifiClient().getProcessGroupClient().replaceProcessGroup(pg.getId(),
importEntity);
+ final String requestId = replaceRequest.getRequest().getRequestId();
+
+ waitFor(() -> {
+ try {
+ final ProcessGroupReplaceRequestEntity req =
+
getNifiClient().getProcessGroupClient().getProcessGroupReplaceRequest(pg.getId(),
requestId);
+ return req != null && req.getRequest().isComplete();
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+
+ final ProcessGroupReplaceRequestEntity finalRequest =
+
getNifiClient().getProcessGroupClient().getProcessGroupReplaceRequest(pg.getId(),
requestId);
+ assertNull(finalRequest.getRequest().getFailureReason(),
+ "Replace failed:
%s".formatted(finalRequest.getRequest().getFailureReason()));
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/FlowDefinitionExportImportStateIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/FlowDefinitionExportImportStateIT.java
new file mode 100644
index 00000000000..8335788f866
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/FlowDefinitionExportImportStateIT.java
@@ -0,0 +1,484 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.tests.system.pg;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.nifi.components.state.Scope;
+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.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
+import org.apache.nifi.web.api.dto.StateEntryDTO;
+import org.apache.nifi.web.api.entity.ComponentStateEntity;
+import org.apache.nifi.web.api.entity.ControllerServiceEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupImportEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupReplaceRequestEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class FlowDefinitionExportImportStateIT extends NiFiSystemIT {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper()
+ .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
false);
+
+ @Test
+ public void testExportWithoutStateFlagHasNoComponentState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st1-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
false, false, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ for (final VersionedProcessor vp :
snapshot.getFlowContents().getProcessors()) {
+ assertNull(vp.getComponentState(), "componentState should be null
when includeComponentState=false");
+ }
+ }
+
+ @Test
+ public void testExportWithStateFlagHasBothScopes() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st2-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "StatefulCountProcessor");
+ assertNotNull(proc);
+
+ final VersionedComponentState state = proc.getComponentState();
+ assertNotNull(state, "componentState should be present when
includeComponentState=true");
+ assertNotNull(state.getLocalNodeStates(), "localNodeStates should be
present for StatefulCountProcessor");
+ assertEquals(1, state.getLocalNodeStates().size());
+ final VersionedNodeState nodeState = state.getLocalNodeStates().get(0);
+ assertNotNull(nodeState);
+ assertNotNull(nodeState.getState().get("count"));
+ assertNull(state.getClusterState(), "clusterState should not be
exported on a standalone node since the cluster state provider is not enabled");
+ }
+
+ @Test
+ public void testExportWithStateWhileRunningReturns409() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+
+ try {
+ final File exportFile = new File("target/st3-export.json");
+
getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(), false,
true, exportFile);
+ throw new AssertionError("Expected export to fail when processors
are running");
+ } catch (final NiFiClientException e) {
+ assertNotNull(e.getMessage());
+ } finally {
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+ }
+ }
+
+ @Test
+ public void testExportWithStateWhileControllerServiceEnabledReturns409()
throws NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ControllerServiceEntity service =
getClientUtil().createControllerService("StandardCountService", pg.getId());
+ getClientUtil().enableControllerService(service);
+
+ try {
+ final File exportFile = new File("target/st4-export.json");
+
getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(), false,
true, exportFile);
+ throw new AssertionError("Expected export to fail when controller
services are enabled");
+ } catch (final NiFiClientException e) {
+ assertNotNull(e.getMessage());
+ } finally {
+ getClientUtil().disableControllerService(service);
+ }
+ }
+
+ @Test
+ public void
testExportWithStateFlagNonStatefulProcessorHasNoComponentState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st5-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor statefulVersioned =
findProcessorByType(snapshot.getFlowContents(), "StatefulCountProcessor");
+ final VersionedProcessor terminateVersioned =
findProcessorByType(snapshot.getFlowContents(), "TerminateFlowFile");
+
+ assertNotNull(statefulVersioned.getComponentState(), "Stateful
processor should have componentState");
+ assertNull(terminateVersioned.getComponentState(), "Non-stateful
processor should not have componentState");
+ }
+
+ @Test
+ public void testExportDefaultNoParamHasNoState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st6-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
false, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ for (final VersionedProcessor vp :
snapshot.getFlowContents().getProcessors()) {
+ assertNull(vp.getComponentState(), "componentState should be null
by default");
+ }
+ }
+
+ @Test
+ public void testStandaloneRoundTripViaUploadLocalState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final Map<String, String> originalLocalState =
getProcessorState(stateful.getId(), Scope.LOCAL);
+ assertNotNull(originalLocalState.get("count"));
+
+ final File exportFile = new File("target/st7-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ emptyQueuesAndDeleteProcessGroup(pg);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedGroup", 0.0, 0.0);
+
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(uploaded.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+
+ final Map<String, String> importedLocalState =
getProcessorState(importedProcessor.getId(), Scope.LOCAL);
+ assertEquals(originalLocalState.get("count"),
importedLocalState.get("count"), "Local state count should match after
round-trip");
+ }
+
+ @Test
+ public void testReplaceRejectsFlowDefinitionWithComponentState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st8-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final NiFiClientException exception =
assertThrows(NiFiClientException.class, () -> replaceProcessGroup(pg,
snapshot));
+ assertTrue(exception.getMessage().contains("component state"),
+ "Expected rejection message about component state but got: " +
exception.getMessage());
+ }
+
+ @Test
+ public void testImportFlowWithoutComponentStateHasNoState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st9-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, false, exportFile);
+
+ emptyQueuesAndDeleteProcessGroup(pg);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedGroup", 0.0, 0.0);
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(uploaded.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+
+ final Map<String, String> importedLocalState =
getProcessorState(importedProcessor.getId(), Scope.LOCAL);
+ assertTrue(importedLocalState.isEmpty(), "State should be empty when
imported without componentState");
+ }
+
+ @Test
+ public void testNestedProcessGroupsStateExportedRecursively() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity parent =
getClientUtil().createProcessGroup("ParentGroup", "root");
+ final ProcessGroupEntity child =
getClientUtil().createProcessGroup("ChildGroup", parent.getId());
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", child.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", child.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final Map<String, String> originalLocalState =
getProcessorState(stateful.getId(), Scope.LOCAL);
+
+ final File exportFile = new File("target/st10-export.json");
+
getNifiClient().getProcessGroupClient().exportProcessGroup(parent.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessGroup childGroup =
snapshot.getFlowContents().getProcessGroups().iterator().next();
+ final VersionedProcessor nestedProcessor =
findProcessorByType(childGroup, "StatefulCountProcessor");
+ assertNotNull(nestedProcessor);
+ assertNotNull(nestedProcessor.getComponentState(), "Nested processor
should have componentState");
+
assertNotNull(nestedProcessor.getComponentState().getLocalNodeStates(), "Nested
processor should have localNodeStates");
+
+ emptyQueuesAndDeleteProcessGroup(parent);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedParent", 0.0, 0.0);
+ final ProcessGroupEntity importedChild =
getNifiClient().getFlowClient().getProcessGroup(uploaded.getId())
+
.getProcessGroupFlow().getFlow().getProcessGroups().iterator().next();
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(importedChild.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+
+ assertEquals(originalLocalState.get("count"),
getProcessorState(importedProcessor.getId(), Scope.LOCAL).get("count"));
+ }
+
+ @Test
+ public void testMultipleStatefulProcessorsEachGetsOwnState() throws
NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity gen1 =
getClientUtil().createProcessor("GenerateFlowFile", pg.getId());
+ final ProcessorEntity gen2 =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(gen1, terminate, "success");
+ getClientUtil().createConnection(gen2, terminate, "success");
+
+ getClientUtil().startProcessor(gen1);
+ getClientUtil().startProcessor(gen2);
+ waitForStatePopulated(gen1.getId(), Scope.LOCAL);
+ waitForStatePopulated(gen2.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(gen1);
+ getClientUtil().stopProcessor(gen2);
+ getClientUtil().waitForStoppedProcessor(gen1.getId());
+ getClientUtil().waitForStoppedProcessor(gen2.getId());
+
+ final Map<String, String> state1 = getProcessorState(gen1.getId(),
Scope.LOCAL);
+ final Map<String, String> state2 = getProcessorState(gen2.getId(),
Scope.LOCAL);
+
+ final File exportFile = new File("target/st11-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ emptyQueuesAndDeleteProcessGroup(pg);
+
+ final ProcessGroupEntity uploaded =
getNifiClient().getProcessGroupClient().upload("root", exportFile,
"ImportedGroup", 0.0, 0.0);
+ final ProcessorEntity imported1 =
findProcessorByTypeInGroup(uploaded.getId(), "GenerateFlowFile");
+ final ProcessorEntity imported2 =
findProcessorByTypeInGroup(uploaded.getId(), "StatefulCountProcessor");
+
+ assertEquals(state1.get("count"), getProcessorState(imported1.getId(),
Scope.LOCAL).get("count"));
+ assertEquals(state2.get("count"), getProcessorState(imported2.getId(),
Scope.LOCAL).get("count"));
+ }
+
+ @Test
+ public void testExportWithStateEmptyStateNeverRan() throws
NiFiClientException, IOException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+
+ final File exportFile = new File("target/st23-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ final VersionedProcessor proc =
findProcessorByType(snapshot.getFlowContents(), "StatefulCountProcessor");
+ assertNull(proc.getComponentState(), "componentState should be null
when processor has never run");
+ }
+
+ @Test
+ public void testExportWithStateNoStatefulComponents() throws
NiFiClientException, IOException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+
+ final File exportFile = new File("target/st24-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, true, exportFile);
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ for (final VersionedProcessor vp :
snapshot.getFlowContents().getProcessors()) {
+ assertNull(vp.getComponentState(), "Non-stateful processor should
have null componentState");
+ }
+ }
+
+ @Test
+ public void testReplaceSucceedsWhenFlowExportedWithoutComponentState()
throws NiFiClientException, IOException, InterruptedException {
+ final ProcessGroupEntity pg =
getClientUtil().createProcessGroup("TestGroup", "root");
+ final ProcessorEntity stateful =
getClientUtil().createProcessor("StatefulCountProcessor", pg.getId());
+ final ProcessorEntity terminate =
getClientUtil().createProcessor("TerminateFlowFile", pg.getId());
+ getClientUtil().createConnection(stateful, terminate, "success");
+
+ getClientUtil().startProcessor(stateful);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ getClientUtil().stopProcessor(stateful);
+ getClientUtil().waitForStoppedProcessor(stateful.getId());
+
+ final File exportFile = new File("target/st27-export.json");
+ getNifiClient().getProcessGroupClient().exportProcessGroup(pg.getId(),
true, false, exportFile);
+
+ getNifiClient().getProcessGroupClient().emptyQueues(pg.getId());
+ waitFor(() -> {
+ try {
+ return
getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId())
+
.getStatus().getAggregateSnapshot().getQueuedCount().equals("0");
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+
+ final RegisteredFlowSnapshot snapshot = MAPPER.readValue(exportFile,
RegisteredFlowSnapshot.class);
+ replaceProcessGroup(pg, snapshot);
+
+ final ProcessorEntity importedProcessor =
findProcessorByTypeInGroup(pg.getId(), "StatefulCountProcessor");
+ assertNotNull(importedProcessor);
+ }
+
+ private void waitForStatePopulated(final String processorId, final Scope
scope) throws InterruptedException {
+ waitFor(() -> {
+ try {
+ final Map<String, String> state =
getProcessorState(processorId, scope);
+ return state.get("count") != null;
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+ }
+
+ private Map<String, String> getProcessorState(final String processorId,
final Scope scope) throws NiFiClientException, IOException {
+ final ComponentStateEntity stateEntity =
getNifiClient().getProcessorClient().getProcessorState(processorId);
+ final ComponentStateDTO componentState =
stateEntity.getComponentState();
+ final Map<String, String> result = new HashMap<>();
+
+ switch (scope) {
+ case LOCAL:
+ if (componentState != null && componentState.getLocalState()
!= null && componentState.getLocalState().getState() != null) {
+ for (final StateEntryDTO entry :
componentState.getLocalState().getState()) {
+ result.put(entry.getKey(), entry.getValue());
+ }
+ }
+ break;
+ case CLUSTER:
+ if (componentState != null && componentState.getClusterState()
!= null && componentState.getClusterState().getState() != null) {
+ for (final StateEntryDTO entry :
componentState.getClusterState().getState()) {
+ result.put(entry.getKey(), entry.getValue());
+ }
+ }
+ break;
+ }
+ return result;
+ }
+
+ private VersionedProcessor findProcessorByType(final VersionedProcessGroup
group, final String typeSuffix) {
+ if (group.getProcessors() != null) {
+ for (final VersionedProcessor vp : group.getProcessors()) {
+ if (vp.getType() != null && vp.getType().endsWith(typeSuffix))
{
+ return vp;
+ }
+ }
+ }
+ return null;
+ }
+
+ private ProcessorEntity findProcessorByTypeInGroup(final String groupId,
final String typeSuffix) throws NiFiClientException, IOException {
+ return getNifiClient().getFlowClient().getProcessGroup(groupId)
+ .getProcessGroupFlow().getFlow().getProcessors().stream()
+ .filter(pe -> pe.getComponent().getType().endsWith(typeSuffix))
+ .findFirst().orElse(null);
+ }
+
+ private void emptyQueuesAndDeleteProcessGroup(final ProcessGroupEntity pg)
throws NiFiClientException, IOException, InterruptedException {
+ getNifiClient().getProcessGroupClient().emptyQueues(pg.getId());
+ waitFor(() -> {
+ try {
+ return
getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId())
+
.getStatus().getAggregateSnapshot().getQueuedCount().equals("0");
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+ final ProcessGroupEntity refreshed =
getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId());
+ refreshed.setDisconnectedNodeAcknowledged(true);
+ getNifiClient().getProcessGroupClient().deleteProcessGroup(refreshed);
+ }
+
+ private void replaceProcessGroup(final ProcessGroupEntity pg, final
RegisteredFlowSnapshot snapshot) throws NiFiClientException, IOException,
InterruptedException {
+ final ProcessGroupImportEntity importEntity = new
ProcessGroupImportEntity();
+ importEntity.setVersionedFlowSnapshot(snapshot);
+
importEntity.setProcessGroupRevision(getNifiClient().getProcessGroupClient().getProcessGroup(pg.getId()).getRevision());
+
+ final ProcessGroupReplaceRequestEntity replaceRequest =
+
getNifiClient().getProcessGroupClient().replaceProcessGroup(pg.getId(),
importEntity);
+ final String requestId = replaceRequest.getRequest().getRequestId();
+
+ waitFor(() -> {
+ try {
+ final ProcessGroupReplaceRequestEntity req =
+
getNifiClient().getProcessGroupClient().getProcessGroupReplaceRequest(pg.getId(),
requestId);
+ return req != null && req.getRequest().isComplete();
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+
+ final ProcessGroupReplaceRequestEntity finalRequest =
+
getNifiClient().getProcessGroupClient().getProcessGroupReplaceRequest(pg.getId(),
requestId);
+ assertNull(finalRequest.getRequest().getFailureReason(),
+ "Replace failed:
%s".formatted(finalRequest.getRequest().getFailureReason()));
+ }
+}
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ProcessGroupClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ProcessGroupClient.java
index 2d530eff05b..a34c27e3d86 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ProcessGroupClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/ProcessGroupClient.java
@@ -68,6 +68,8 @@ public interface ProcessGroupClient {
File exportProcessGroup(String processGroupId, boolean
includeReferencedServices, File outputFile) throws NiFiClientException,
IOException;
+ File exportProcessGroup(String processGroupId, boolean
includeReferencedServices, boolean includeComponentState, File outputFile)
throws NiFiClientException, IOException;
+
DropRequestEntity emptyQueues(String processGroupId) throws
NiFiClientException, IOException;
DropRequestEntity getEmptyQueuesRequest(String processGroupId, String
requestId) throws NiFiClientException, IOException;
diff --git
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyProcessGroupClient.java
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyProcessGroupClient.java
index 19798f8015e..f5e294e0030 100644
---
a/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyProcessGroupClient.java
+++
b/nifi-toolkit/nifi-toolkit-client/src/main/java/org/apache/nifi/toolkit/client/impl/JerseyProcessGroupClient.java
@@ -274,6 +274,12 @@ public class JerseyProcessGroupClient extends
AbstractJerseyClient implements Pr
@Override
public File exportProcessGroup(final String processGroupId, final boolean
includeReferencedServices, final File outputFile) throws NiFiClientException,
IOException {
+ return exportProcessGroup(processGroupId, includeReferencedServices,
false, outputFile);
+ }
+
+ @Override
+ public File exportProcessGroup(final String processGroupId, final boolean
includeReferencedServices, final boolean includeComponentState,
+ final File outputFile) throws
NiFiClientException, IOException {
if (StringUtils.isBlank(processGroupId)) {
throw new IllegalArgumentException("Process group id cannot be
null or blank");
}
@@ -282,7 +288,8 @@ public class JerseyProcessGroupClient extends
AbstractJerseyClient implements Pr
final WebTarget target = processGroupsTarget
.path("{id}/download")
.resolveTemplate("id", processGroupId)
- .queryParam("includeReferencedServices",
includeReferencedServices);
+ .queryParam("includeReferencedServices",
includeReferencedServices)
+ .queryParam("includeComponentState",
includeComponentState);
final Response response = getRequestBuilder(target)
.accept(MediaType.APPLICATION_JSON)
