You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "bobpaulin (via GitHub)" <gi...@apache.org> on 2026/04/22 12:20:52 UTC
Re: [PR] NIFI-15671 - Flow import/export with stateful components state [nifi]
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