This is an automated email from the ASF dual-hosted git repository.
pvillard 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 57956a62703 NIFI-15774: Consolidate checks for whether or not a
component must be updated when applying flow differences; ensure on flow change
rollback we resolve compatible bundles; added system tests to verify behavior;
minor updates to cursor rules to avoid violations of code style and best
practices
57956a62703 is described below
commit 57956a6270305335ea48643c1113ccb92e376858
Author: Mark Payne <[email protected]>
AuthorDate: Tue Mar 31 17:45:55 2026 -0400
NIFI-15774: Consolidate checks for whether or not a component must be
updated when applying flow differences; ensure on flow change rollback we
resolve compatible bundles; added system tests to verify behavior; minor
updates to cursor rules to avoid violations of code style and best practices
This closes #11085.
Signed-off-by: Pierre Villard <[email protected]>
---
.cursor/rules/best-practices.mdc | 9 +
.cursor/rules/building.mdc | 29 +--
.cursor/rules/code-style.mdc | 14 --
.cursor/rules/comments.mdc | 14 ++
.cursor/rules/ending-conditions.mdc | 14 --
.cursor/rules/extension-development.mdc | 14 --
.cursor/rules/extension-testing.mdc | 14 --
.cursor/rules/framework-testing.mdc | 20 +-
.cursor/rules/persona.mdc | 14 --
.cursor/rules/testing-standards.mdc | 14 --
.../StandardVersionedComponentSynchronizer.java | 9 +-
.../java/org/apache/nifi/util/BundleUtils.java | 24 ++-
.../apache/nifi/util/FlowDifferenceFilters.java | 31 +++
.../nifi/util/TestFlowDifferenceFilters.java | 51 +++++
.../apache/nifi/web/StandardNiFiServiceFacade.java | 7 +-
.../apache/nifi/web/api/FlowUpdateResource.java | 8 +
.../VersionChangeWithIncompatibleBundleIT.java | 236 +++++++++++++++++++++
pom.xml | 1 +
18 files changed, 395 insertions(+), 128 deletions(-)
diff --git a/.cursor/rules/best-practices.mdc b/.cursor/rules/best-practices.mdc
new file mode 100644
index 00000000000..42a18cbbe07
--- /dev/null
+++ b/.cursor/rules/best-practices.mdc
@@ -0,0 +1,9 @@
+---
+description: General coding best practices for maintainable NiFi changes
+alwaysApply: true
+---
+
+# Best Practices
+
+- Reuse expensive, thread-safe helpers instead of constructing them ad hoc in
methods or tests.
+- For example, when a class needs an `ObjectMapper`, prefer a shared field
such as `private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();`
unless a special configuration requires a dedicated instance.
diff --git a/.cursor/rules/building.mdc b/.cursor/rules/building.mdc
index 7ec9fabd47c..aa3c9d69b38 100644
--- a/.cursor/rules/building.mdc
+++ b/.cursor/rules/building.mdc
@@ -2,20 +2,6 @@
description: Maven build instructions for the NiFi codebase
alwaysApply: true
---
-<!--
- 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.
--->
# Building
@@ -30,3 +16,18 @@ codebase, but only after you have attempted to build the
relevant modules yourse
It is important not to run `mvn clean` at the root level or at the
`nifi-assembly` level without
the user's express permission, as this may delete a running instance of NiFi,
causing permanent
loss of flows and configuration.
+
+When preparing to run system tests, packaged runtime artifacts must be
refreshed using
+`mvn clean install` on the relevant packaging modules. Do not rely on `mvn
install` or
+`mvn package` alone for system tests, as stale NARs or assemblies may be
reused.
+
+If a change affects code packaged into a NAR, rebuild the packaging module
that delivers that
+code before running system tests. For example, changes under
+`nifi-framework-bundle/nifi-framework/nifi-web/...` require rebuilding
+`nifi-framework-bundle/nifi-server-nar`, because the runtime uses
`nifi-server-nar` to deliver
+the updated web application classes.
+
+Before running tests in `nifi-system-tests/nifi-system-test-suite`, refresh
its runtime assembly
+with `mvn clean install` for that module as well. The system test runtime uses
+`target/nifi-lib-assembly`, which is populated in `pre-integration-test` and
later phases, so
+running `mvn test` or `mvn package` alone may leave stale runtime contents in
place.
diff --git a/.cursor/rules/code-style.mdc b/.cursor/rules/code-style.mdc
index 5432070ecf7..ca7e9cfed5a 100644
--- a/.cursor/rules/code-style.mdc
+++ b/.cursor/rules/code-style.mdc
@@ -3,20 +3,6 @@ description: Java code style conventions for the NiFi codebase
globs: "**/*.java"
alwaysApply: false
---
-<!--
- 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.
--->
# Code Style
diff --git a/.cursor/rules/comments.mdc b/.cursor/rules/comments.mdc
new file mode 100644
index 00000000000..50060f3bf2f
--- /dev/null
+++ b/.cursor/rules/comments.mdc
@@ -0,0 +1,14 @@
+---
+description: Keep source and test comments durable and code-focused
+alwaysApply: true
+---
+
+# Comment Guidance
+
+Write comments as durable explanations of the code or test contract itself.
+
+- Do not write temporary commentary such as "without the fix", "with the fix",
"before this change", or "after this change".
+- Do not describe a code path relative to the current patch or bugfix timeline.
+- In tests, describe the scenario, setup, and behavior being verified, not how
the test behaved before or after a fix.
+- Do not add redundant assertion messages that simply restate the assertion or
expected value. Use an assertion message only when it adds unique diagnostic
context that the assertion itself does not already provide.
+- If historical context matters, put it in commit messages, pull request
descriptions, or issue tracking, not in source comments.
diff --git a/.cursor/rules/ending-conditions.mdc
b/.cursor/rules/ending-conditions.mdc
index 8242bb34005..aa2b607801a 100644
--- a/.cursor/rules/ending-conditions.mdc
+++ b/.cursor/rules/ending-conditions.mdc
@@ -2,20 +2,6 @@
description: Task completion checklist that must be verified before
considering any task done
alwaysApply: true
---
-<!--
- 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.
--->
# Ending Conditions
diff --git a/.cursor/rules/extension-development.mdc
b/.cursor/rules/extension-development.mdc
index 614bc79b09e..74e53f34d5f 100644
--- a/.cursor/rules/extension-development.mdc
+++ b/.cursor/rules/extension-development.mdc
@@ -2,20 +2,6 @@
description: Development patterns for NiFi extensions (Processors, Controller
Services, Connectors). Covers Property Descriptors, Relationships, and common
patterns.
alwaysApply: false
---
-<!--
- 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.
--->
# Extension Development
diff --git a/.cursor/rules/extension-testing.mdc
b/.cursor/rules/extension-testing.mdc
index e5143336f38..97692657289 100644
--- a/.cursor/rules/extension-testing.mdc
+++ b/.cursor/rules/extension-testing.mdc
@@ -2,20 +2,6 @@
description: Testing guidance for NiFi extensions (Processors, Controller
Services, Connectors). Covers nifi-mock and TestRunner usage.
alwaysApply: false
---
-<!--
- 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.
--->
# Extension Testing
diff --git a/.cursor/rules/framework-testing.mdc
b/.cursor/rules/framework-testing.mdc
index 3343b0d0102..4b5c5e8dfaa 100644
--- a/.cursor/rules/framework-testing.mdc
+++ b/.cursor/rules/framework-testing.mdc
@@ -2,20 +2,6 @@
description: Testing guidance for NiFi framework code (not extensions). Covers
when to use unit, integration, and system tests for framework classes.
alwaysApply: false
---
-<!--
- 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.
--->
# Framework Testing
@@ -53,6 +39,12 @@ Good candidates for system tests include changes to
`ProcessScheduler`, `Process
`ControllerServiceNode`, `FlowController`, `FlowManager`, how Parameters are
handled, flow
synchronization, the repositories, etc.
+When a system test switches the client to a specific cluster node using
`switchClientToNode(...)`
+and then needs to verify the state of that node specifically, the request must
bypass request
+replication by passing `NiFiSystemIT.DO_NOT_REPLICATE` to the client call.
Otherwise, the request
+may be replicated or merged across the cluster and will not reliably reflect
the state of the
+selected node.
+
## Escalation
Any unit test that ends up requiring a large number of mocks is a good
candidate for an
diff --git a/.cursor/rules/persona.mdc b/.cursor/rules/persona.mdc
index d741efe5d99..3e4a66ed645 100644
--- a/.cursor/rules/persona.mdc
+++ b/.cursor/rules/persona.mdc
@@ -2,20 +2,6 @@
description: AI persona and general approach for working on the Apache NiFi
codebase
alwaysApply: true
---
-<!--
- 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.
--->
# AI Persona
diff --git a/.cursor/rules/testing-standards.mdc
b/.cursor/rules/testing-standards.mdc
index 676b099dd70..6cb98f4853c 100644
--- a/.cursor/rules/testing-standards.mdc
+++ b/.cursor/rules/testing-standards.mdc
@@ -3,20 +3,6 @@ description: Shared test coding conventions for all NiFi
automated tests (framew
globs: "**/src/test/**/*.java"
alwaysApply: false
---
-<!--
- 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.
--->
# Testing Standards
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 424b4782c2d..092d2f7e7bb 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
@@ -328,10 +328,7 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
setSynchronizationOptions(options);
for (final FlowDifference diff : flowComparison.getDifferences()) {
- if
(FlowDifferenceFilters.isPropertyMissingFromGhostComponent(diff,
context.getFlowManager())) {
- continue;
- }
- if (FlowDifferenceFilters.isScheduledStateNew(diff)) {
+ if (!FlowDifferenceFilters.isComponentUpdateRequired(diff,
versionedExternalFlow.getFlowContents(), context.getFlowManager())) {
continue;
}
@@ -355,10 +352,6 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
}
}
- if (diff.getDifferenceType() == DifferenceType.POSITION_CHANGED) {
- continue;
- }
-
final VersionedComponent component = diff.getComponentA() == null
? diff.getComponentB() : diff.getComponentA();
updatedVersionedComponentIds.add(component.getIdentifier());
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
index 554f5a5d94d..cc1fd83eacb 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/BundleUtils.java
@@ -26,6 +26,8 @@ import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.nar.NarClassLoadersHolder;
import org.apache.nifi.nar.PythonBundle;
import org.apache.nifi.web.api.dto.BundleDTO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
@@ -35,6 +37,8 @@ import java.util.Optional;
* Utility class for Bundles.
*/
public final class BundleUtils {
+ private static final Logger logger =
LoggerFactory.getLogger(BundleUtils.class);
+
static Optional<BundleCoordinate> findOptionalBundleForType(final
ExtensionManager extensionManager, final String type, final Bundle
frameworkBundle) {
final List<Bundle> bundles = extensionManager.getBundles(type);
if (bundles.size() == 1) {
@@ -246,16 +250,32 @@ public final class BundleUtils {
}
public static void discoverCompatibleBundle(final ExtensionManager
extensionManager, final ParameterProviderReference parameterProviderReference) {
- final BundleDTO dto =
createBundleDto(parameterProviderReference.getBundle());
+ final org.apache.nifi.flow.Bundle originalBundle =
parameterProviderReference.getBundle();
+ final BundleDTO dto = createBundleDto(originalBundle);
final BundleCoordinate coordinate =
getOptionalCompatibleBundle(extensionManager,
parameterProviderReference.getType(), dto).orElse(
new BundleCoordinate(dto.getGroup(), dto.getArtifact(),
dto.getVersion()));
+
+ if (!coordinate.getVersion().equals(originalBundle.getVersion())) {
+ logger.info("Discovered compatible bundle for [{}] - original
bundle [{}:{}:{}] mapped to compatible bundle [{}:{}:{}]",
parameterProviderReference.getType(),
+ originalBundle.getGroup(), originalBundle.getArtifact(),
originalBundle.getVersion(),
+ coordinate.getGroup(), coordinate.getId(),
coordinate.getVersion());
+ }
+
parameterProviderReference.setBundle(createBundle(coordinate));
}
public static void discoverCompatibleBundle(final ExtensionManager
extensionManager, final VersionedConfigurableExtension extension) {
- final BundleDTO dto = createBundleDto(extension.getBundle());
+ final org.apache.nifi.flow.Bundle originalBundle =
extension.getBundle();
+ final BundleDTO dto = createBundleDto(originalBundle);
final BundleCoordinate coordinate =
getOptionalCompatibleBundle(extensionManager, extension.getType(), dto).orElse(
new BundleCoordinate(dto.getGroup(), dto.getArtifact(),
dto.getVersion()));
+
+ if (!coordinate.getVersion().equals(originalBundle.getVersion())) {
+ logger.info("Discovered compatible bundle for [{}] - original
bundle [{}:{}:{}] mapped to compatible bundle [{}:{}:{}]", extension.getType(),
+ originalBundle.getGroup(), originalBundle.getArtifact(),
originalBundle.getVersion(),
+ coordinate.getGroup(), coordinate.getId(),
coordinate.getVersion());
+ }
+
extension.setBundle(createBundle(coordinate));
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/FlowDifferenceFilters.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/FlowDifferenceFilters.java
index 1d952864f5a..e34922b75c0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/FlowDifferenceFilters.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/FlowDifferenceFilters.java
@@ -107,6 +107,37 @@ public class FlowDifferenceFilters {
|| isPropertyAddedFromMigration(difference, flowManager);
}
+ /**
+ * Determines whether a Flow Difference represents a change that requires
stopping and updating the affected component. Returns false for differences
+ * that are benign with respect to the running state of a component -- for
example, a new property added with a default value, or a position change.
+ *
+ * This method is used by the synchronizer to decide which components to
update and by the affected-components calculation to decide which components
+ * to stop. Both callers must agree; otherwise a component may be updated
without being stopped first, causing failures.
+ *
+ * Note: {@link #isLocalScheduleStateChange} is intentionally not included
here. During a versioned flow update the
+ * {@code RetainExistingStateComponentScheduler} already preserves local
scheduled-state, so including it would be harmless but redundant. During
+ * cluster reconnection, however, scheduled-state differences must be
applied so the reconnecting node matches the cluster, and filtering them out
+ * would prevent the synchronizer from starting or stopping components as
needed. Callers that build an "affected components" set for version
+ * changes should apply {@link #isLocalScheduleStateChange} as a separate
filter to avoid flagging components whose only difference is a
+ * user-initiated state change.
+ *
+ * @param difference the Flow Difference to evaluate
+ * @param proposedContents the proposed VersionedProcessGroup from the
target flow snapshot
+ * @param flowManager the Flow Manager
+ * @return true if the difference requires a component update, false if it
can be safely ignored
+ */
+ public static boolean isComponentUpdateRequired(final FlowDifference
difference, final VersionedProcessGroup proposedContents, final FlowManager
flowManager) {
+ if (difference.getDifferenceType() == DifferenceType.POSITION_CHANGED)
{
+ return false;
+ }
+
+ return !isPropertyMissingFromGhostComponent(difference, flowManager)
+ && !isScheduledStateNew(difference)
+ && !isNewPropertyWithDefaultValue(difference, flowManager)
+ && !isNewRelationshipAutoTerminatedAndDefaulted(difference,
proposedContents, flowManager)
+ && !isStaticPropertyRemoved(difference, flowManager);
+ }
+
public static boolean isBundleChange(final FlowDifference difference) {
return difference.getDifferenceType() == DifferenceType.BUNDLE_CHANGED;
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java
index 6466a082c72..72ae6d5d3aa 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/util/TestFlowDifferenceFilters.java
@@ -740,6 +740,57 @@ public class TestFlowDifferenceFilters {
assertFalse(FlowDifferenceFilters.isEnvironmentalChange(difference,
null, flowManager));
}
+ @Test
+ public void testIsComponentUpdateRequiredForPositionChange() {
+ final FlowManager flowManager = Mockito.mock(FlowManager.class);
+ final VersionedProcessor processorA = new VersionedProcessor();
+ final VersionedProcessor processorB = new VersionedProcessor();
+
+ final StandardFlowDifference positionDiff = new StandardFlowDifference(
+ DifferenceType.POSITION_CHANGED, processorA, processorB, null,
null, "Position changed");
+
+
assertFalse(FlowDifferenceFilters.isComponentUpdateRequired(positionDiff, null,
flowManager));
+ }
+
+ @Test
+ public void testIsComponentUpdateRequiredForScheduledStateNew() {
+ final FlowManager flowManager = Mockito.mock(FlowManager.class);
+ final VersionedProcessor processorA = new VersionedProcessor();
+ final VersionedProcessor processorB = new VersionedProcessor();
+ processorA.setScheduledState(null);
+
processorB.setScheduledState(org.apache.nifi.flow.ScheduledState.RUNNING);
+
+ final StandardFlowDifference scheduledStateDiff = new
StandardFlowDifference(
+ DifferenceType.SCHEDULED_STATE_CHANGED, processorA, processorB,
+ processorA.getScheduledState(),
processorB.getScheduledState(), "");
+
+
assertFalse(FlowDifferenceFilters.isComponentUpdateRequired(scheduledStateDiff,
null, flowManager));
+ }
+
+ @Test
+ public void testIsComponentUpdateRequiredForPropertyChange() {
+ final FlowManager flowManager = Mockito.mock(FlowManager.class);
+ final VersionedProcessor processorA = new VersionedProcessor();
+ final VersionedProcessor processorB = new VersionedProcessor();
+
+ final StandardFlowDifference propertyDiff = new StandardFlowDifference(
+ DifferenceType.PROPERTY_CHANGED, processorA, processorB,
"oldValue", "newValue", "Property changed");
+
+
assertTrue(FlowDifferenceFilters.isComponentUpdateRequired(propertyDiff, null,
flowManager));
+ }
+
+ @Test
+ public void testIsComponentUpdateRequiredForLocalScheduleStateChange() {
+ final FlowManager flowManager = Mockito.mock(FlowManager.class);
+ final VersionedProcessor processorA = new VersionedProcessor();
+ final VersionedProcessor processorB = new VersionedProcessor();
+
+ final StandardFlowDifference scheduledStateDiff = new
StandardFlowDifference(
+ DifferenceType.SCHEDULED_STATE_CHANGED, processorA,
processorB, "STOPPED", "RUNNING", "");
+
+
assertTrue(FlowDifferenceFilters.isComponentUpdateRequired(scheduledStateDiff,
null, flowManager));
+ }
+
@DynamicProperty(name = "Dynamic Property", value = "Value", description =
"Allows dynamic properties")
private static class DynamicAnnotationProcessor extends AbstractProcessor {
@Override
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 f9057f97249..3257d81385c 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
@@ -6350,13 +6350,8 @@ public class StandardNiFiServiceFacade implements
NiFiServiceFacade {
.filter(difference -> difference.getDifferenceType() !=
DifferenceType.COMPONENT_ADDED) // components that are added are not components
that will be affected in the local flow.
.filter(FlowDifferenceFilters.FILTER_ADDED_REMOVED_REMOTE_PORTS)
.filter(difference -> difference.getComponentA() != null) // a
difference that would not affect a local component
- .filter(diff ->
!FlowDifferenceFilters.isNewPropertyWithDefaultValue(diff, flowManager))
- .filter(diff ->
!FlowDifferenceFilters.isNewRelationshipAutoTerminatedAndDefaulted(diff,
proposedFlow.getContents(), flowManager))
- .filter(diff ->
!FlowDifferenceFilters.isScheduledStateNew(diff))
+ .filter(diff ->
FlowDifferenceFilters.isComponentUpdateRequired(diff,
proposedFlow.getContents(), flowManager))
.filter(diff ->
!FlowDifferenceFilters.isLocalScheduleStateChange(diff))
- .filter(diff ->
!FlowDifferenceFilters.isPropertyMissingFromGhostComponent(diff, flowManager))
- .filter(diff ->
!FlowDifferenceFilters.isStaticPropertyRemoved(diff, flowManager))
- .filter(difference -> difference.getDifferenceType() !=
DifferenceType.POSITION_CHANGED)
.map(difference -> {
final VersionedComponent localComponent =
difference.getComponentA();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java
index 36e6907b701..72257509465 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/FlowUpdateResource.java
@@ -398,6 +398,14 @@ public abstract class FlowUpdateResource<T extends
ProcessGroupDescriptorEntity,
} else {
final FlowSnapshotContainer originalFlowSnapshotContainer =
serviceFacade.getVersionedFlowSnapshot(vciEntity.getVersionControlInformation(),
true);
originalFlowSnapshot =
originalFlowSnapshotContainer.getFlowSnapshot();
+
+ // Resolve compatible bundles, inherited controller services, and
parameter providers for the rollback snapshot before any
+ // replication occurs, ensuring that all nodes in the cluster
receive the same resolved references.
+
serviceFacade.discoverCompatibleBundles(originalFlowSnapshot.getFlowContents());
+
serviceFacade.discoverCompatibleBundles(originalFlowSnapshot.getParameterProviders());
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+
serviceFacade.resolveInheritedControllerServices(originalFlowSnapshotContainer,
groupId, user);
+ serviceFacade.resolveParameterProviders(originalFlowSnapshot,
user);
}
try {
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/VersionChangeWithIncompatibleBundleIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/VersionChangeWithIncompatibleBundleIT.java
new file mode 100644
index 00000000000..636a40c60be
--- /dev/null
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/registry/VersionChangeWithIncompatibleBundleIT.java
@@ -0,0 +1,236 @@
+/*
+ * 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.registry;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.nifi.tests.system.NiFiClientUtil;
+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.ProcessorDTO;
+import org.apache.nifi.web.api.dto.flow.FlowDTO;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.FlowRegistryClientEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.apache.nifi.web.api.entity.VersionControlInformationEntity;
+import org.apache.nifi.web.api.entity.VersionedFlowUpdateRequestEntity;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Verifies that changing a flow version in a cluster correctly resolves
incompatible bundle
+ * versions on all nodes. This reproduces a scenario where a flow registry
contains snapshots
+ * with bundle versions that don't match the installed NARs (e.g., after a
NiFi upgrade).
+ * The coordinator resolves bundles via discoverCompatibleBundles(), and this
test verifies
+ * the resolution propagates to all cluster nodes during replication.
+ */
+public class VersionChangeWithIncompatibleBundleIT extends NiFiSystemIT {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private static final String TEST_FLOWS_BUCKET = "test-flows";
+ private static final String INCOMPATIBLE_VERSION = "0.0.0-DOES-NOT-EXIST";
+
+ @Override
+ public NiFiInstanceFactory getInstanceFactory() {
+ return createTwoNodeInstanceFactory();
+ }
+
+ /**
+ * Tests that when changing to a flow version whose registry snapshot
contains bundle
+ * versions not installed on the cluster, the bundles are resolved to
compatible versions
+ * and no ghost processors are created on any node.
+ *
+ * Steps:
+ * 1. Create a flow with two processors and save as v1 and v2 (both with
correct bundles)
+ * 2. Switch back to v1
+ * 3. Modify v2's snapshot.json on disk to use a non-existent bundle
version, simulating
+ * what happens when a flow was registered with a different NiFi version
+ * 4. Change to v2 — the coordinator should resolve the incompatible
bundles
+ * 5. Verify that all processors on both cluster nodes have the correct
bundle version
+ * and are not ghosted
+ */
+ @Test
+ public void testVersionChangeResolvesIncompatibleBundleAcrossCluster()
throws NiFiClientException, IOException, InterruptedException {
+ final File storageDir = new File("target/flowRegistryStorage/" +
getTestName().replaceAll("\\(.*?\\)", ""));
+ Files.createDirectories(storageDir.toPath());
+ final FlowRegistryClientEntity clientEntity =
registerClient(storageDir);
+ final NiFiClientUtil util = getClientUtil();
+
+ final ProcessGroupEntity group = util.createProcessGroup("Parent",
"root");
+ final ProcessorEntity generate =
util.createProcessor("GenerateFlowFile", group.getId());
+ final ProcessorEntity terminate =
util.createProcessor("TerminateFlowFile", group.getId());
+ util.createConnection(generate, terminate, "success");
+
+ final VersionControlInformationEntity vci =
util.startVersionControl(group, clientEntity, TEST_FLOWS_BUCKET, "Parent");
+ final String flowId = vci.getVersionControlInformation().getFlowId();
+
+ util.updateProcessorProperties(generate,
Collections.singletonMap("Text", "Hello World"));
+ util.saveFlowVersion(group, clientEntity, vci);
+
+ util.changeFlowVersion(group.getId(), "1");
+ util.assertFlowStaleAndUnmodified(group.getId());
+
+ modifySnapshotBundleVersions(storageDir, flowId, "2",
INCOMPATIBLE_VERSION);
+
+ final VersionedFlowUpdateRequestEntity updateResult =
util.changeFlowVersion(group.getId(), "2");
+ assertNull(updateResult.getRequest().getFailureReason());
+ util.assertFlowUpToDate(group.getId());
+
+ verifyProcessorsResolved(group.getId());
+
+ switchClientToNode(2);
+ verifyProcessorsResolved(group.getId());
+ }
+
+ /**
+ * Tests that when a version change fails and triggers a rollback, the
rollback snapshot
+ * (fetched from the flow registry) also has its bundles resolved. This
reproduces the
+ * production scenario where:
+ *
+ * 1. NiFi is upgraded, so all registry snapshots have old bundle versions
+ * 2. A version change is initiated but fails (e.g., connection has data)
+ * 3. The rollback fetches the ORIGINAL version's snapshot from the
registry
+ * 4. That snapshot also has incompatible bundles that must be resolved
+ */
+ @Test
+ public void testRollbackResolvesIncompatibleBundles() throws
NiFiClientException, IOException, InterruptedException {
+ final File storageDir = new File("target/flowRegistryStorage/" +
getTestName().replaceAll("\\(.*?\\)", ""));
+ Files.createDirectories(storageDir.toPath());
+ final FlowRegistryClientEntity clientEntity =
registerClient(storageDir);
+ final NiFiClientUtil util = getClientUtil();
+
+ final ProcessGroupEntity group = util.createProcessGroup("Parent",
"root");
+ final ProcessorEntity generate =
util.createProcessor("GenerateFlowFile", group.getId());
+ final ProcessorEntity terminate =
util.createProcessor("TerminateFlowFile", group.getId());
+ util.createConnection(generate, terminate, "success");
+
+ final VersionControlInformationEntity vci =
util.startVersionControl(group, clientEntity, TEST_FLOWS_BUCKET, "Parent");
+ final String flowId = vci.getVersionControlInformation().getFlowId();
+
+ // v2: remove connection and TerminateFlowFile, auto-terminate
GenerateFlowFile
+ final FlowDTO currentFlow =
getNifiClient().getFlowClient().getProcessGroup(group.getId()).getProcessGroupFlow().getFlow();
+ for (final ConnectionEntity connectionEntity :
currentFlow.getConnections()) {
+
getNifiClient().getConnectionClient().deleteConnection(connectionEntity);
+ }
+
+ getNifiClient().getProcessorClient().deleteProcessor(terminate);
+ util.setAutoTerminatedRelationships(generate, "success");
+ util.saveFlowVersion(group, clientEntity, vci);
+
+ // Go back to v1 so we have the connection (GenerateFlowFile ->
TerminateFlowFile)
+ util.changeFlowVersion(group.getId(), "1");
+ util.assertFlowStaleAndUnmodified(group.getId());
+
+ // Get the newly re-created connection ID (it changes after version
switch)
+ final FlowDTO v1Flow =
getNifiClient().getFlowClient().getProcessGroup(group.getId()).getProcessGroupFlow().getFlow();
+ final String connectionId =
v1Flow.getConnections().iterator().next().getId();
+
+ // Get the re-created GenerateFlowFile processor entity
+ ProcessorEntity generateEntity = null;
+ for (final ProcessorEntity processor : v1Flow.getProcessors()) {
+ if ("GenerateFlowFile".equals(processor.getComponent().getName()))
{
+ generateEntity = processor;
+ break;
+ }
+ }
+
+ assertNotNull(generateEntity, "GenerateFlowFile processor should exist
after reverting to v1");
+
+ // Generate FlowFiles to put data in the connection
+ util.startProcessor(generateEntity);
+ waitForQueueCount(connectionId, getNumberOfNodes());
+ util.stopProcessor(generateEntity);
+
+ // Modify v1's snapshot (the rollback target) to have incompatible
bundles
+ modifySnapshotBundleVersions(storageDir, flowId, "1",
INCOMPATIBLE_VERSION);
+
+ // Change to v2. This should fail because the connection has data and
v2 removes it.
+ // The rollback will fetch v1 from the registry, which now has
incompatible bundles.
+ final VersionedFlowUpdateRequestEntity updateResult =
util.changeFlowVersion(group.getId(), "2", false);
+ assertNotNull(updateResult.getRequest().getFailureReason());
+
+ // After rollback, we should still be at v1 with resolved (non-ghost)
processors
+ util.assertFlowStaleAndUnmodified(group.getId());
+
+ verifyProcessorsResolved(group.getId());
+
+ switchClientToNode(2);
+ verifyProcessorsResolved(group.getId());
+ }
+
+ private void modifySnapshotBundleVersions(final File storageDir, final
String flowId, final String version,
+ final String newBundleVersion)
throws IOException {
+ final File snapshotFile = new File(storageDir, "test-flows/" + flowId
+ "/" + version + "/snapshot.json");
+ final JsonNode root = OBJECT_MAPPER.readTree(snapshotFile);
+
+ modifyBundleVersionsInProcessGroup((ObjectNode)
root.get("flowContents"), newBundleVersion);
+
+
OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValue(snapshotFile, root);
+ }
+
+ private void modifyBundleVersionsInProcessGroup(final ObjectNode
processGroup, final String newVersion) {
+ if (processGroup == null) {
+ return;
+ }
+
+ final JsonNode processors = processGroup.get("processors");
+ if (processors != null && processors.isArray()) {
+ for (final JsonNode processor : processors) {
+ final ObjectNode bundle = (ObjectNode) processor.get("bundle");
+ if (bundle != null) {
+ bundle.put("version", newVersion);
+ }
+ }
+ }
+
+ final JsonNode childGroups = processGroup.get("processGroups");
+ if (childGroups != null && childGroups.isArray()) {
+ for (final JsonNode childGroup : childGroups) {
+ modifyBundleVersionsInProcessGroup((ObjectNode) childGroup,
newVersion);
+ }
+ }
+ }
+
+ private void verifyProcessorsResolved(final String groupId) throws
NiFiClientException, IOException {
+ final FlowDTO flow =
getNifiClient().getFlowClient(DO_NOT_REPLICATE).getProcessGroup(groupId).getProcessGroupFlow().getFlow();
+ final Set<ProcessorEntity> processors = flow.getProcessors();
+
+ assertNotNull(processors);
+ assertFalse(processors.isEmpty());
+
+ for (final ProcessorEntity processorEntity : processors) {
+ final ProcessorDTO processor = processorEntity.getComponent();
+
+ assertFalse(processor.getExtensionMissing());
+ assertEquals(getNiFiVersion(), processor.getBundle().getVersion());
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 6b40e5e26d6..3f32da6948d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1058,6 +1058,7 @@
<exclude>nb-configuration.xml</exclude> <!-- courtesy
excludes for netbeans users -->
<exclude>nbactions.xml</exclude> <!-- courtesy
excludes for netbeans users -->
<exclude>DEPENDENCIES</exclude> <!-- auto generated
file by apache's maven config while building sources.zip -->
+ <exclude>.cursor/rules/**</exclude>
</excludes>
</configuration>
</plugin>
