You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "pvillard31 (via GitHub)" <gi...@apache.org> on 2026/03/09 19:38:13 UTC
[PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
pvillard31 opened a new pull request, #10986:
URL: https://github.com/apache/nifi/pull/10986
# Summary
NIFI-15671 - Flow import/export with stateful components state
This extends the JSON-based "Download flow definition" / "Upload flow definition" workflow to optionally include component state (processors and controller services annotated with `@Stateful`). A new `includeComponentState` query parameter on the download endpoint controls this. The feature does not apply to registry-based flow versioning.
**Key behavioral decisions:**
- **Opt-in via query parameter.** `GET /process-groups/{id}/download?includeComponentState=true` includes state; the default is `false`, preserving backward compatibility. The existing `ObjectMapper` configuration (`FAIL_ON_UNKNOWN_PROPERTIES = false`) means older NiFi versions silently ignore the new field.
- **Never registry versioning.** `FlowMappingOptions.DEFAULT_OPTIONS` (used by the registry path) has `mapComponentState=false`. This is architecturally enforced, not just a convention.
- **Components must be stopped.** Exporting with state requires all processors in the group to be stopped and all controller services disabled. If any are running/enabled, the endpoint returns `409 Conflict`. This guarantees state consistency.
- **Authorization escalation.** Exporting with state requires `WRITE` permission on the process group (accessing state requires write access). Without state, only `READ` is needed.
- **Upload only, not replace.** State is restored only for newly created components during upload. The replace endpoints (`PUT` and `POST .../replace-requests`) reject any flow definition containing component state with `400 Bad Request`, because replacing an existing group produces a mix of new/updated/unchanged components where selective state restoration would be confusing.
- **Cluster LOCAL state handling.** In a cluster, the export request is replicated to all nodes; each contributes its own LOCAL state keyed by ordinal position. The coordinator merges the responses. On import, each node picks up the state entry at its ordinal index. Ordinals are determined by sorting nodes by `address:port`.
- **Cluster topology validation on import.** If the flow definition contains local state from `S` source nodes and the destination cluster has `D` connected nodes: `S <= D` is allowed (extra nodes get no state), `S > D` is rejected with `409 Conflict`.
- **External services have state stripped.** When `includeReferencedServices=true`, controller services from parent groups are included for wiring but their state is set to null — state export is scoped to the target group only.
- **Only `@Stateful` components get state.** Non-stateful components always have null `componentState`, even when `includeComponentState=true`.
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [ ] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [ ] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [ ] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
- [ ] Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status
### Pull Request Formatting
- [ ] Pull Request based on current revision of the `main` branch
- [ ] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [ ] Build completed using `./mvnw clean install -P contrib-check`
- [ ] JDK 21
- [ ] JDK 25
### Licensing
- [ ] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ ] Documentation formatting appears as expected in rendered files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3182235375
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,104 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
return propertyValues;
}
+ private void validateLocalStateTopology(final VersionedProcessGroup proposed) {
+ final int connectedNodeCount = context.getConnectedNodeCount();
Review Comment:
`StandardProcessGroup` now derives both `localNodeOrdinal` and `connectedNodeCount` from `ClusterTopologyProvider`, which is backed by `NodeConnectionState.CONNECTED`, so the count reflects only connected nodes and matches the variable name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3154687830
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java:
##########
@@ -2956,7 +2958,7 @@ public Set<String> getClusterMembers() {
@Override
public Optional<String> getCurrentNode() {
if (isClustered() && getNodeId() != null) {
- return Optional.of(getNodeId().getApiAddress());
+ return Optional.of("%s:%d".formatted(getNodeId().getApiAddress(), getNodeId().getApiPort()));
Review Comment:
This is a risky change, as they break backward compatibility because it changes what is returned by the `NodeTypeProvider`. In particular, this affects `EmbeddedHazelcastCacheManager`.
After this change:
- `m + PORT_SEPARATOR + port` (line 141) becomes `"address:nifiApiPort:hazelcastPort"` — Hazelcast will fail to bind / connect.
- `hazelcastMembers.contains(getCurrentNode().get())` (line 152) compares user-configured hostnames against `"address:port"` and will always return `false`, sending the local node down the client path instead of becoming a member.
- The `customValidate` check at lines 224–236 compares user-configured Hazelcast hosts against getClusterMembers() and will always report them as "not part of the NiFi cluster".
GetAzureEventHub.getClientIdentifier() also folds the value into a client identifier, which is now silently changed. That is less likely to break things but is still a behavioral change.
The export/import feature only needs a stable, unique key for ordinal calculation. Mixing port into a public `NodeTypeProvider` method that other extensions already depend on is the wrong layer to fix it. One option: Introduce a new framework-internal helper that returns `Set<NodeIdentifier>`/sorted IDs and use it only from `StandardProcessGroup` / `StandardNiFiServiceFacade`. Leave `getClusterMembers()` / `getCurrentNode()` alone
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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 {
Review Comment:
This test only re-reads cluster state after import. It does not query each node's local state to confirm Node 0's exported state ends up on Node 0 (and not on Node 1). Add an assertion that uses getProcessorState(... Scope.LOCAL) per node.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,104 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
return propertyValues;
}
+ private void validateLocalStateTopology(final VersionedProcessGroup proposed) {
+ final int connectedNodeCount = context.getConnectedNodeCount();
Review Comment:
`connectedNodeCount` is a misleading name
This made says it compares "S source nodes" against "D connected nodes" (per the PR description), but the `connectedNodeCount` it reads is actually `Math.max(sortedMembers.size(), 1)` — the size of `getClusterMembers()`, which is all members regardless of connection state. So a 5-node cluster with only 2 connected nodes will accept a 4-node export and silently leave two of the four state slices unrestored. We should either rename the field/builder method to `clusterMemberCount` and document, or compute it from connected nodes only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on PR #10986:
URL: https://github.com/apache/nifi/pull/10986#issuecomment-4424232634
Thanks for updating and for putting this together, @pvillard31 ! All looks good to me at this point. The system tests do a pretty good job of verifying basic functionality but I also tested it manually, using curl to export the flow with state. Visually inspected that it looked good, then used curl to import a new PG from it. Verified the state was copied to the new PG. Also then cleared the state on one, and ran both to ensure that they were properly updating independently without issue. All checks out +1 will merge!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 merged PR #10986:
URL: https://github.com/apache/nifi/pull/10986
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on PR #10986:
URL: https://github.com/apache/nifi/pull/10986#issuecomment-4400097378
Thanks @markap14 - I pushed a commit to address your feedback and proceeded with additional testing locally with the latest changes on standalone and clusters with local state and cluster state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "bobpaulin (via GitHub)" <gi...@apache.org>.
bobpaulin commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3120580898
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,102 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
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.debug("StateManagerProvider is not available; skipping state restoration for component {}", componentId);
Review Comment:
This makes sense that not having the state manager provider will prevent us from being able to import state. Not blocking the import make sense since the user may want to proceed with out it. However I would challenge if this should be logged only at debug. Shouldn't the user be notified if they were expecting to import state but it was not? These seems like it would very likely impact the behavior after the flow is started. I'd consider this may be a WARN level event.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,102 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
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.debug("StateManagerProvider is not available; skipping state restoration for component {}", componentId);
+ return;
+ }
+
+ final ConfigurableComponent component = componentNode.getComponent();
+ if (component == null) {
+ LOG.debug("Component {} is not available; skipping state restoration", componentId);
+ return;
+ }
+
+ final Stateful stateful = component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ LOG.debug("Component {} ({}) is not annotated with @Stateful; skipping state restoration", 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.debug("Local node ordinal is not set; skipping local state restoration for component {}", componentId);
Review Comment:
WARN level event?
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,102 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
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.debug("StateManagerProvider is not available; skipping state restoration for component {}", componentId);
+ return;
+ }
+
+ final ConfigurableComponent component = componentNode.getComponent();
+ if (component == null) {
+ LOG.debug("Component {} is not available; skipping state restoration", componentId);
Review Comment:
WARN level event?
##########
nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java:
##########
@@ -4080,6 +4096,102 @@ private Map<String, String> getPropertyValues(final ComponentNode componentNode)
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.debug("StateManagerProvider is not available; skipping state restoration for component {}", componentId);
+ return;
+ }
+
+ final ConfigurableComponent component = componentNode.getComponent();
+ if (component == null) {
+ LOG.debug("Component {} is not available; skipping state restoration", componentId);
+ return;
+ }
+
+ final Stateful stateful = component.getClass().getAnnotation(Stateful.class);
+ if (stateful == null) {
+ LOG.debug("Component {} ({}) is not annotated with @Stateful; skipping state restoration", componentId, component.getClass().getSimpleName());
Review Comment:
WARN level event?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3182238238
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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 {
Review Comment:
Extended `testClusterRoundTripSameTopologyBothScopes` to capture per-node LOCAL state grouped by `clusterNodeId` before export and assert the same map after import, verifying each node's local state lands back on the correct node.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3182230063
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java:
##########
@@ -2956,7 +2958,7 @@ public Set<String> getClusterMembers() {
@Override
public Optional<String> getCurrentNode() {
if (isClustered() && getNodeId() != null) {
- return Optional.of(getNodeId().getApiAddress());
+ return Optional.of("%s:%d".formatted(getNodeId().getApiAddress(), getNodeId().getApiPort()));
Review Comment:
Reverted the changes and added a new framework-internal ClusterTopologyProvider interface implemented by FlowController.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
Posted by "markap14 (via GitHub)" <gi...@apache.org>.
markap14 commented on code in PR #10986:
URL: https://github.com/apache/nifi/pull/10986#discussion_r3202616581
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java:
##########
@@ -323,6 +347,160 @@ public Response exportProcessGroup(
return generateOkResponse(currentVersionedFlowSnapshot).header(HttpHeaders.CONTENT_DISPOSITION, String.format("attachment; filename=\"%s\"", filename)).build();
}
+ /**
+ * Replicates the flow export request to all cluster nodes and merges the LOCAL state from each node
+ * into a single response. Each node returns a RegisteredFlowSnapshot containing only its own LOCAL
+ * state (keyed by its ordinal). This method collects all node responses and combines the localNodeStates
+ * maps so the exported flow contains LOCAL state from every node.
+ *
+ * @return the merged response containing LOCAL state from all nodes
+ */
+ private Response replicateAndMergeFlowExport() {
Review Comment:
Perhaps I'm missing something, but I think we should register an `EndpointResponseMerger` for `GET /process-groups/<uuid>/download`. This is the typical pattern that we follow to handle merging the responses. A merger would also let the framework drain unread response streams consistently.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/pg/ClusterFlowDefinitionExportImportStateIT.java:
##########
@@ -0,0 +1,350 @@
+/*
+ * 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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);
+ waitForStatePopulated(stateful.getId(), Scope.LOCAL);
+ 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);
+
+ 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 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 {
Review Comment:
`waitForStatePopulated` and `getProcessorState` — the wait polls a flattened map and returns as soon as either node has populated count. The tests then assert two local-state slices are present at lines 104 (testClusterExportCapturesLocalStateFromBothNodes), 135 (testClusterExportCapturesBothScopes), and 155 / 172 (testClusterRoundTripSameTopologyBothScopes). If the second node has not yet executed onTrigger by the time the wait returns, those assertions will flake. The wait should explicitly verify both ordinals have state (e.g., getProcessorLocalStatesByNode(...).size() == 2) for the round-trip case.
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java:
##########
@@ -5935,24 +5982,58 @@ private Set<String> getAllSubGroups(ProcessGroup processGroup) {
return result;
}
+ /**
+ * Computes the ordinal index of the local node among connected cluster nodes.
+ * Nodes are sorted deterministically by apiAddress. In standalone mode, returns 0.
+ *
+ * @return the ordinal index of the local node, or 0 if not clustered
+ */
+ private int computeLocalNodeOrdinal() {
Review Comment:
This duplicates the logic added in FlowController.getLocalNodeOrdinal() in the same PR. The new ClusterTopologyProvider interface was introduced for exactly this purpose; the service facade should depend on it rather than reimplementing the ordinal calculation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org