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 <[email protected]>
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
*/
