You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by pv...@apache.org on 2026/03/29 15:38:42 UTC
(nifi) branch main updated: NIFI-15744: Working Flow must be synchronized after being discarded (#11064)
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 81a5932b163 NIFI-15744: Working Flow must be synchronized after being discarded (#11064)
81a5932b163 is described below
commit 81a5932b163741cd7f0970576ee2240611510890
Author: Bob Paulin <bo...@bobpaulin.com>
AuthorDate: Sun Mar 29 10:38:36 2026 -0500
NIFI-15744: Working Flow must be synchronized after being discarded (#11064)
---
.../connector/StandardConnectorNode.java | 27 +++++
.../connector/TestStandardConnectorNode.java | 123 ++++++++++++++++++++-
2 files changed, 149 insertions(+), 1 deletion(-)
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
index d2f53dcdd43..02675f47299 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
@@ -861,6 +861,33 @@ public class StandardConnectorNode implements ConnectorNode {
connectorDetails.getComponentLog(), activeFlowContext.getConfigurationContext(), activeFlowContext.getBundle());
getComponentLog().info("Working Flow Context has been recreated");
+
+ synchronizeWorkingFlowParameters();
+ }
+
+ /**
+ * Re-triggers {@link Connector#onConfigurationStepConfigured} for every configured step in
+ * the working flow context. This ensures that flow parameters derived from the configuration
+ * (e.g., resolved asset paths) are fresh, even when the working context was just recreated
+ * from the active flow whose parameter values may be stale.
+ *
+ * <p>Skipped when {@code initializationContext} is {@code null} because the connector has
+ * not yet been initialized and there is no flow to update.</p>
+ */
+ private void synchronizeWorkingFlowParameters() {
+ if (initializationContext == null) {
+ return;
+ }
+
+ final ConnectorConfiguration config = workingFlowContext.getConfigurationContext().toConnectorConfiguration();
+ for (final NamedStepConfiguration stepConfig : config.getNamedStepConfigurations()) {
+ try {
+ setConfiguration(stepConfig.stepName(), stepConfig.configuration(), true);
+ } catch (final Exception e) {
+ logger.warn("Failed to synchronize working flow parameters for step [{}] of {}",
+ stepConfig.stepName(), this, e);
+ }
+ }
}
@Override
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
index d306ac45aa8..c637a22b02f 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
@@ -409,6 +409,102 @@ public class TestStandardConnectorNode {
assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("configurationStep1"));
}
+ @Test
+ public void testDiscardWorkingConfigurationCallsOnStepConfigured() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("prop1", "value1")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.discardWorkingConfiguration();
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ }
+
+ @Test
+ public void testApplyUpdateCallsOnStepConfiguredForAllConfiguredSteps() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("prop1", "value1")));
+ connectorNode.setConfiguration("step2", createStepConfiguration(Map.of("prop2", "value2")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("prop1", "value2")));
+ connectorNode.applyUpdate();
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step2"));
+ }
+
+ @Test
+ public void testDiscardWorkingConfigurationCallsOnStepConfiguredForMultipleSteps() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("prop1", "value1")));
+ connectorNode.setConfiguration("step2", createStepConfiguration(Map.of("prop2", "value2")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.discardWorkingConfiguration();
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step2"));
+ }
+
+ @Test
+ public void testApplyUpdateWithNoChangesCallsOnStepConfigured() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("prop1", "value1")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.applyUpdate();
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ }
+
+ @Test
+ public void testSynchronizeWorkingFlowParametersContinuesOnStepFailure() throws FlowUpdateException {
+ final FailingStepConnector failingStepConnector = new FailingStepConnector("failingStep");
+ final StandardConnectorNode connectorNode = createConnectorNode(failingStepConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("failingStep", createStepConfiguration(Map.of("prop1", "value1")));
+ connectorNode.setConfiguration("successStep", createStepConfiguration(Map.of("prop2", "value2")));
+ connectorNode.applyUpdate();
+
+ failingStepConnector.reset();
+ failingStepConnector.setFailOnStep(true);
+
+ connectorNode.discardWorkingConfiguration();
+
+ assertTrue(failingStepConnector.wasOnPropertyGroupConfiguredCalled("successStep"));
+ }
+
@Test
public void testVerifyConfigurationStepSurfacesInvalidValidationResults() throws FlowUpdateException {
final ValidationFailingConnector validationFailingConnector = new ValidationFailingConnector();
@@ -647,7 +743,7 @@ public class TestStandardConnectorNode {
}
@Override
- protected void onStepConfigured(final String stepName, final FlowContext workingContext) {
+ protected void onStepConfigured(final String stepName, final FlowContext workingContext) throws FlowUpdateException {
onConfigurationStepConfiguredCalls.add(stepName);
}
@@ -665,6 +761,31 @@ public class TestStandardConnectorNode {
}
}
+ /**
+ * Test connector that extends TrackingConnector to throw from onStepConfigured for a specific step name.
+ * Used to verify that synchronization continues even when one step fails.
+ */
+ private static class FailingStepConnector extends TrackingConnector {
+ private final String failingStepName;
+ private boolean failOnStep = false;
+
+ public FailingStepConnector(final String failingStepName) {
+ this.failingStepName = failingStepName;
+ }
+
+ public void setFailOnStep(final boolean failOnStep) {
+ this.failOnStep = failOnStep;
+ }
+
+ @Override
+ protected void onStepConfigured(final String stepName, final FlowContext workingContext) throws FlowUpdateException {
+ super.onStepConfigured(stepName, workingContext);
+ if (failOnStep && failingStepName.equals(stepName)) {
+ throw new FlowUpdateException("Simulated failure for step " + stepName);
+ }
+ }
+ }
+
/**
* Test connector that returns invalid validation results from validateConfigurationStep
*/