You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2026/05/14 16:41:16 UTC
(nifi) branch main updated: NIFI-15930 Inherited Controller Services before Parameter Providers for Secrets (#11252)
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory 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 c9ec26629e3 NIFI-15930 Inherited Controller Services before Parameter Providers for Secrets (#11252)
c9ec26629e3 is described below
commit c9ec26629e34dd1da5fe539af285ddf8a4b6f089
Author: Bob Paulin <bo...@bobpaulin.com>
AuthorDate: Thu May 14 11:41:09 2026 -0500
NIFI-15930 Inherited Controller Services before Parameter Providers for Secrets (#11252)
- Moved inherit Controller Services before inherit Parameter Providers
- Enable Working Context to apply any changes in Connector Property Values and Resolved values to the Parameter Context of the Working Flow Context
- Warn when a parameter provider is not found
Signed-off-by: David Handermann <ex...@apache.org>
---
.../secrets/ParameterProviderSecretsManager.java | 88 +++++++--
.../TestParameterProviderSecretsManager.java | 71 +++++++
.../nifi/components/connector/ConnectorNode.java | 21 ++
.../connector/StandardConnectorNode.java | 42 ++--
.../connector/StandardConnectorRepository.java | 29 ++-
.../serialization/VersionedFlowSynchronizer.java | 11 +-
.../connector/TestStandardConnectorNode.java | 62 +++++-
.../connector/TestStandardConnectorRepository.java | 219 ++++++++++++++++++++-
8 files changed, 481 insertions(+), 62 deletions(-)
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
index c9cc7bef209..37d7d5c5755 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/components/connector/secrets/ParameterProviderSecretsManager.java
@@ -50,6 +50,14 @@ public class ParameterProviderSecretsManager implements SecretsManager {
private Duration cacheDuration;
private final Map<String, CachedSecret> secretCache = new ConcurrentHashMap<>();
+ // Per-ParameterProvider-id deduplication for the WARN log emitted whenever a SecretReference
+ // resolution is skipped because its backing provider is not VALID. An entry is present for any
+ // provider id we have already surfaced via WARN; the value is the status that was warned about
+ // so the same WARN is not repeated until the status either changes or the provider returns to
+ // VALID. Entries for providers that no longer exist in the flow are pruned at the top of
+ // getSecretProviders() so the map stays bounded across many flow synchronizations.
+ private final Map<String, ValidationStatus> lastWarnedStatus = new ConcurrentHashMap<>();
+
private record CachedSecret(Secret secret, long timestampNanos) {
}
@@ -79,17 +87,41 @@ public class ParameterProviderSecretsManager implements SecretsManager {
@Override
public Set<SecretProvider> getSecretProviders() {
+ final Set<ParameterProviderNode> parameterProviderNodes = flowManager.getAllParameterProviders();
+
+ // Drop dedup entries for any Parameter Provider that no longer exists in the flow so the map
+ // does not grow unbounded when providers are created INVALID and deleted before going VALID.
+ final Set<String> currentProviderIds = new HashSet<>();
+ for (final ParameterProviderNode parameterProviderNode : parameterProviderNodes) {
+ final String providerId = parameterProviderNode.getIdentifier();
+ if (providerId != null) {
+ currentProviderIds.add(providerId);
+ }
+ }
+ lastWarnedStatus.keySet().retainAll(currentProviderIds);
+
final Set<SecretProvider> providers = new HashSet<>();
- for (final ParameterProviderNode parameterProviderNode : flowManager.getAllParameterProviders()) {
+ for (final ParameterProviderNode parameterProviderNode : parameterProviderNodes) {
ValidationStatus validationStatus = parameterProviderNode.getValidationStatus();
if (validationStatus != ValidationStatus.VALID) {
validationStatus = parameterProviderNode.performValidation();
}
if (validationStatus != ValidationStatus.VALID) {
- logger.debug("Will not use Parameter Provider {} as a Secret Provider because it is not valid", parameterProviderNode.getName());
+ logSkippedInvalidProvider(parameterProviderNode, validationStatus);
continue;
}
+ // Clear the WARN-dedup state for this provider and surface a paired INFO so each
+ // earlier WARN has an explicit recovery line operators can correlate against.
+ final String providerId = parameterProviderNode.getIdentifier();
+ if (providerId != null) {
+ final ValidationStatus priorWarnedStatus = lastWarnedStatus.remove(providerId);
+ if (priorWarnedStatus != null) {
+ logger.info("Parameter Provider [{}] (id={}) returned to VALID after being logged as {};"
+ + " SecretReferences backed by this provider will resolve again",
+ parameterProviderNode.getName(), providerId, priorWarnedStatus);
+ }
+ }
providers.add(new ParameterProviderSecretProvider(parameterProviderNode));
}
@@ -143,26 +175,23 @@ public class ParameterProviderSecretsManager implements SecretsManager {
private Map<SecretReference, Secret> fetchSecretsWithoutCache(final Set<SecretReference> secretReferences) {
final Set<SecretProvider> providers = getSecretProviders();
- // Partition secret references by Provider
+ // Partition secret references by Provider. References whose provider is non-VALID or absent
+ // are recorded with a null Secret so that callers receive an explicit entry for every input.
final Map<SecretProvider, Set<SecretReference>> referencesByProvider = new HashMap<>();
+ final Map<SecretReference, Secret> secrets = new HashMap<>();
for (final SecretReference secretReference : secretReferences) {
final SecretProvider provider = findProvider(secretReference, providers);
- referencesByProvider.computeIfAbsent(provider, k -> new HashSet<>()).add(secretReference);
+ if (provider == null) {
+ secrets.put(secretReference, null);
+ } else {
+ referencesByProvider.computeIfAbsent(provider, k -> new HashSet<>()).add(secretReference);
+ }
}
- final Map<SecretReference, Secret> secrets = new HashMap<>();
for (final Map.Entry<SecretProvider, Set<SecretReference>> entry : referencesByProvider.entrySet()) {
final SecretProvider provider = entry.getKey();
final Set<SecretReference> references = entry.getValue();
- // If no provider found, be sure to map to a null Secret rather than skipping
- if (provider == null) {
- for (final SecretReference secretReference : references) {
- secrets.put(secretReference, null);
- }
- continue;
- }
-
final List<String> secretNames = references.stream()
.map(SecretReference::getFullyQualifiedName)
.filter(Objects::nonNull)
@@ -192,6 +221,7 @@ public class ParameterProviderSecretsManager implements SecretsManager {
// Partition references into cache hits vs. misses that need fetching
final Map<SecretProvider, Set<SecretReference>> uncachedByProvider = new HashMap<>();
+
for (final SecretReference secretReference : secretReferences) {
final String fqn = secretReference.getFullyQualifiedName();
@@ -205,7 +235,11 @@ public class ParameterProviderSecretsManager implements SecretsManager {
}
final SecretProvider provider = findProvider(secretReference, providers);
- uncachedByProvider.computeIfAbsent(provider, k -> new HashSet<>()).add(secretReference);
+ if (provider == null) {
+ results.put(secretReference, null);
+ } else {
+ uncachedByProvider.computeIfAbsent(provider, k -> new HashSet<>()).add(secretReference);
+ }
}
// Batch fetch uncached secrets grouped by provider
@@ -213,13 +247,6 @@ public class ParameterProviderSecretsManager implements SecretsManager {
final SecretProvider provider = entry.getKey();
final Set<SecretReference> references = entry.getValue();
- if (provider == null) {
- for (final SecretReference secretReference : references) {
- results.put(secretReference, null);
- }
- continue;
- }
-
final List<String> secretNames = references.stream()
.map(SecretReference::getFullyQualifiedName)
.filter(Objects::nonNull)
@@ -266,6 +293,25 @@ public class ParameterProviderSecretsManager implements SecretsManager {
}
}
+ private void logSkippedInvalidProvider(final ParameterProviderNode parameterProviderNode, final ValidationStatus status) {
+ final String providerId = parameterProviderNode.getIdentifier();
+ if (providerId == null) {
+ return;
+ }
+
+ // Treat a null status (e.g., a mock or transient lookup failure) as INVALID for deduplication
+ // purposes so the tracked status is always well-defined.
+ final ValidationStatus effectiveStatus = status == null ? ValidationStatus.INVALID : status;
+ final ValidationStatus priorStatus = lastWarnedStatus.put(providerId, effectiveStatus);
+ if (priorStatus == effectiveStatus) {
+ return;
+ }
+
+ logger.warn("Skipping Parameter Provider [{}] (id={}) as a Secret Provider because its current validation status is {}; "
+ + "SecretReferences backed by this provider will resolve to null until it returns to VALID",
+ parameterProviderNode.getName(), providerId, effectiveStatus);
+ }
+
private SecretProvider findProvider(final SecretReference secretReference, final Set<SecretProvider> providers) {
// Search first by Provider ID, if it's provided.
final String providerId = secretReference.getProviderId();
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
index 6d5545c9fca..7ba784df077 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/components/connector/secrets/TestParameterProviderSecretsManager.java
@@ -575,5 +575,76 @@ public class TestParameterProviderSecretsManager {
manager.initialize(initContext);
return manager;
}
+
+ @Test
+ public void testGetSecretsReturnsNullValueWhenProviderIsInvalid() {
+ final FlowManager flowManager = mock(FlowManager.class);
+ final ParameterProviderNode invalidProvider = createMockedParameterProviderNode(PROVIDER_1_ID, PROVIDER_1_NAME, GROUP_1_NAME,
+ ValidationStatus.INVALID, createParameter(SECRET_1_NAME, SECRET_1_DESCRIPTION, SECRET_1_VALUE));
+ // performValidation should also return INVALID so the provider is consistently filtered out.
+ when(invalidProvider.performValidation()).thenReturn(ValidationStatus.INVALID);
+ when(flowManager.getAllParameterProviders()).thenReturn(Set.of(invalidProvider));
+
+ final ParameterProviderSecretsManager manager = new ParameterProviderSecretsManager();
+ manager.initialize(new StandardSecretsManagerInitializationContext(flowManager));
+
+ final SecretReference reference = createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_1_NAME);
+
+ final Map<SecretReference, Secret> results = manager.getSecrets(Set.of(reference));
+
+ assertEquals(1, results.size());
+ assertTrue(results.containsKey(reference));
+ assertNull(results.get(reference));
+ }
+
+ @Test
+ public void testGetSecretsResolvesReferenceOnceProviderTransitionsToValid() {
+ final FlowManager flowManager = mock(FlowManager.class);
+ final ParameterProviderNode flippingProvider = createMockedParameterProviderNode(PROVIDER_1_ID, PROVIDER_1_NAME, GROUP_1_NAME,
+ ValidationStatus.INVALID, createParameter(SECRET_1_NAME, SECRET_1_DESCRIPTION, SECRET_1_VALUE));
+ when(flippingProvider.performValidation()).thenReturn(ValidationStatus.INVALID);
+ when(flowManager.getAllParameterProviders()).thenReturn(Set.of(flippingProvider));
+
+ final ParameterProviderSecretsManager manager = new ParameterProviderSecretsManager();
+ manager.initialize(new StandardSecretsManagerInitializationContext(flowManager,
+ Map.of(NiFiProperties.SECRETS_MANAGER_CACHE_DURATION, DEFAULT_CACHE_DURATION)));
+
+ final SecretReference reference = createSecretReference(PROVIDER_1_ID, PROVIDER_1_NAME, SECRET_1_NAME);
+
+ // While the provider is INVALID the secret is unresolvable.
+ assertNull(manager.getSecrets(Set.of(reference)).get(reference));
+
+ // Once the provider transitions to VALID a follow-up call resolves the value. The cache only
+ // stores non-null secrets, so the prior unresolved attempt does not block this re-attempt.
+ when(flippingProvider.getValidationStatus()).thenReturn(ValidationStatus.VALID);
+ when(flippingProvider.performValidation()).thenReturn(ValidationStatus.VALID);
+
+ final Secret resolved = manager.getSecrets(Set.of(reference)).get(reference);
+ assertNotNull(resolved);
+ assertEquals(SECRET_1_VALUE, resolved.getValue());
+ }
+
+ @Test
+ public void testGetSecretProvidersFiltersConsistentlyAcrossValidationStatusTransitions() {
+ final FlowManager flowManager = mock(FlowManager.class);
+ final ParameterProviderNode provider = createMockedParameterProviderNode(PROVIDER_1_ID, PROVIDER_1_NAME, GROUP_1_NAME,
+ ValidationStatus.INVALID, createParameter(SECRET_1_NAME, SECRET_1_DESCRIPTION, SECRET_1_VALUE));
+ when(provider.performValidation()).thenReturn(ValidationStatus.INVALID);
+ when(flowManager.getAllParameterProviders()).thenReturn(Set.of(provider));
+
+ final ParameterProviderSecretsManager manager = new ParameterProviderSecretsManager();
+ manager.initialize(new StandardSecretsManagerInitializationContext(flowManager));
+
+ assertTrue(manager.getSecretProviders().isEmpty());
+ assertTrue(manager.getSecretProviders().isEmpty());
+
+ when(provider.getValidationStatus()).thenReturn(ValidationStatus.VALID);
+ when(provider.performValidation()).thenReturn(ValidationStatus.VALID);
+ assertEquals(1, manager.getSecretProviders().size());
+
+ when(provider.getValidationStatus()).thenReturn(ValidationStatus.INVALID);
+ when(provider.performValidation()).thenReturn(ValidationStatus.INVALID);
+ assertTrue(manager.getSecretProviders().isEmpty());
+ }
}
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
index bec52c28754..08819d38857 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorNode.java
@@ -242,6 +242,27 @@ public interface ConnectorNode extends ComponentAuthorizable, VersionedComponent
*/
void setConfiguration(String configurationStepName, StepConfiguration configuration) throws FlowUpdateException;
+ /**
+ * Replaces the configuration of the named step on the working flow context with the given configuration.
+ * Unlike {@link #setConfiguration(String, StepConfiguration)}, which merges the incoming properties with
+ * any existing properties for the step, this method treats the supplied configuration as the authoritative
+ * full state for the step: any property not present in {@code configuration} is removed from the step.
+ *
+ * <p>If applying the configuration changes either the raw or the resolved property values, the Connector
+ * is notified via {@link Connector#onConfigurationStepConfigured} so the embedded flow and Parameter
+ * Context can be brought up to date. If nothing changed, no notification is performed.</p>
+ *
+ * <p>Intended for use by the framework when reconciling the working flow context against an external
+ * {@link ConnectorConfigurationProvider}, whose view is treated as authoritative. This method should
+ * only be invoked via the ConnectorRepository.</p>
+ *
+ * @param configurationStepName the name of the configuration step being replaced
+ * (must match one of the names returned by {@link Connector#getConfigurationSteps()})
+ * @param configuration the full configuration for the given configuration step
+ * @throws FlowUpdateException if unable to apply the configuration changes
+ */
+ void replaceWorkingConfiguration(String configurationStepName, StepConfiguration configuration) throws FlowUpdateException;
+
void transitionStateForUpdating();
void prepareForUpdate() throws FlowUpdateException;
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 a647af3b683..b9659ebb486 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
@@ -333,18 +333,30 @@ public class StandardConnectorNode implements ConnectorNode {
}
private void setConfiguration(final String stepName, final StepConfiguration configuration, final boolean forceOnConfigurationStepConfigured) throws FlowUpdateException {
- // Update properties and check if the configuration changed.
final ConfigurationUpdateResult updateResult = workingFlowContext.getConfigurationContext().setProperties(stepName, configuration);
if (updateResult == ConfigurationUpdateResult.NO_CHANGES && !forceOnConfigurationStepConfigured) {
return;
}
+ notifyStepConfigured(stepName);
+ }
- // If there were changes, trigger Processor to be notified of the change.
+ @Override
+ public void replaceWorkingConfiguration(final String stepName, final StepConfiguration configuration) throws FlowUpdateException {
+ // The configuration provider's view is authoritative: any property absent from the provided
+ // configuration is removed from the step.
+ final ConfigurationUpdateResult updateResult = workingFlowContext.getConfigurationContext().replaceProperties(stepName, configuration);
+ if (updateResult == ConfigurationUpdateResult.NO_CHANGES) {
+ return;
+ }
+ notifyStepConfigured(stepName);
+ }
+
+ private void notifyStepConfigured(final String stepName) throws FlowUpdateException {
final Connector connector = connectorDetails.getConnector();
try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, connector.getClass(), getIdentifier())) {
logger.debug("Notifying {} of configuration change for configuration step {}", this, stepName);
connector.onConfigurationStepConfigured(stepName, workingFlowContext);
- logger.debug("Successfully set configuration for step {} on {}", stepName, this);
+ logger.debug("Successfully notified {} of configuration change for step {}", this, stepName);
} catch (final FlowUpdateException e) {
throw e;
} catch (final Exception e) {
@@ -861,34 +873,26 @@ public class StandardConnectorNode implements ConnectorNode {
workingFlowContext = flowContextFactory.createWorkingFlowContext(identifier,
connectorDetails.getComponentLog(), activeFlowContext.getConfigurationContext(), activeFlowContext.getBundle());
- getComponentLog().info("Working Flow Context has been recreated");
+ getComponentLog().info("Working Flow Context has been set");
- 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() {
+ // Re-fire onConfigurationStepConfigured for every step so flow parameters derived from the
+ // configuration (e.g., resolved asset paths, secret values) are refreshed against the new
+ // working context. Step failures are logged so the remaining steps can still be refreshed.
+ // Skipped before the connector has been initialized because there is no flow to update yet.
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 {}",
+ logger.warn("Failed to refresh resolved configuration for step [{}] of {}",
stepConfig.stepName(), this, e);
}
}
+
+ getComponentLog().info("Working Flow Context configuration has been refreshed");
}
@Override
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
index 3e2d8ede2ea..bdb527fecc8 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java
@@ -813,15 +813,26 @@ public class StandardConnectorRepository implements ConnectorRepository {
}
final List<VersionedConfigurationStep> workingFlowConfiguration = config.getWorkingFlowConfiguration();
- if (workingFlowConfiguration != null) {
- // Enrich provider-sourced SECRET_REFERENCE values with providerId before they are
- // converted into the in-memory ConnectorValueReference graph.
- resolveSecretReferencesFromProvider(workingFlowConfiguration);
-
- final MutableConnectorConfigurationContext workingConfigContext = connector.getWorkingFlowContext().getConfigurationContext();
- for (final VersionedConfigurationStep step : workingFlowConfiguration) {
- final StepConfiguration stepConfiguration = toStepConfiguration(step);
- workingConfigContext.replaceProperties(step.getName(), stepConfiguration);
+
+ if (workingFlowConfiguration == null) {
+ return;
+ }
+
+ // Enrich provider-sourced SECRET_REFERENCE values with providerId before they are
+ // converted into the in-memory ConnectorValueReference graph.
+ resolveSecretReferencesFromProvider(workingFlowConfiguration);
+
+ // Replace each step's working configuration on the connector. Routing through the connector
+ // (rather than touching the configuration context directly) ensures it is notified via
+ // onConfigurationStepConfigured when raw or resolved property values changed, so the embedded
+ // flow's Parameter Context picks up new values (e.g., rotated secrets) without an explicit save.
+ for (final VersionedConfigurationStep step : workingFlowConfiguration) {
+ final StepConfiguration stepConfiguration = toStepConfiguration(step);
+ try {
+ connector.replaceWorkingConfiguration(step.getName(), stepConfiguration);
+ } catch (final Exception e) {
+ logger.warn("Failed to replace working configuration for step [{}] on {} during sync from provider; continuing with remaining steps",
+ step.getName(), connector, e);
}
}
}
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
index 52fc732be7a..ba409ee7d44 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
@@ -417,15 +417,12 @@ public class VersionedFlowSynchronizer implements FlowSynchronizer {
versionedExternalFlow.setParameterContexts(versionedParameterContextMap);
versionedExternalFlow.setFlowContents(versionedFlow.getRootGroup());
- // Inherit Parameter Providers and Connectors first. Because Connectors are a bit different, in that updates could result in Exceptions being thrown,
- // due to the fact that they manipulate the flow, and changes can be aborted, we handle them first. This way, if there's any Exception,
- // we can fail before updating parts of the flow that are not managed by Connectors.
- // Because Connectors may depend on Parameter Providers, we need to ensure that we inherit Parameter Providers first.
+ // Inherit root-level Controller Services first so that Parameter Providers backed by
+ // them are VALID when Connectors begin resolving SecretReferences and so that any other
+ // controller-level component that references a root CS sees it in its target state.
+ inheritControllerServices(controller, versionedFlow, affectedComponentSet);
inheritParameterProviders(controller, versionedFlow, affectedComponentSet);
inheritConnectors(controller, versionedFlow);
-
- // Inherit controller-level components.
- inheritControllerServices(controller, versionedFlow, affectedComponentSet);
inheritParameterContexts(controller, versionedFlow);
inheritReportingTasks(controller, versionedFlow, affectedComponentSet);
inheritFlowAnalysisRules(controller, versionedFlow, affectedComponentSet);
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 365c19dbff8..00e136a6635 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
@@ -491,7 +491,67 @@ public class TestStandardConnectorNode {
}
@Test
- public void testSynchronizeWorkingFlowParametersContinuesOnStepFailure() throws FlowUpdateException {
+ public void testReplaceWorkingConfigurationReplacesPropertiesAndFiresWhenChanged() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("propA", "valueA", "propB", "valueB")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.replaceWorkingConfiguration("step1", createStepConfiguration(Map.of("propA", "newA")));
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ final ConnectorConfiguration workingConfig = connectorNode.getWorkingFlowContext().getConfigurationContext().toConnectorConfiguration();
+ final NamedStepConfiguration namedStep = workingConfig.getNamedStepConfigurations().iterator().next();
+ assertEquals("step1", namedStep.stepName());
+ assertEquals(Map.of("propA", new StringLiteralValue("newA")), namedStep.configuration().getPropertyValues());
+ }
+
+ @Test
+ public void testReplaceWorkingConfigurationDoesNotFireWhenUnchanged() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("propA", "valueA")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ connectorNode.replaceWorkingConfiguration("step1", createStepConfiguration(Map.of("propA", "valueA")));
+
+ assertFalse(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ }
+
+ @Test
+ public void testDiscardWorkingConfigurationFiresOnConfiguredForEveryWorkingStep() throws FlowUpdateException {
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createConnectorNode(trackingConnector);
+
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration("step1", createStepConfiguration(Map.of("propA", "valueA")));
+ connectorNode.setConfiguration("step2", createStepConfiguration(Map.of("propB", "valueB")));
+ connectorNode.applyUpdate();
+
+ trackingConnector.reset();
+
+ // Recreating the working flow context from the active flow must fire onConfigurationStepConfigured
+ // for every working configuration step so that flow parameters derived from the configuration
+ // (resolved asset paths, secrets, etc.) are refreshed.
+ connectorNode.discardWorkingConfiguration();
+
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step1"));
+ assertTrue(trackingConnector.wasOnPropertyGroupConfiguredCalled("step2"));
+ }
+
+ @Test
+ public void testDiscardWorkingConfigurationContinuesOnStepFailure() throws FlowUpdateException {
final FailingStepConnector failingStepConnector = new FailingStepConnector("failingStep");
final StandardConnectorNode connectorNode = createConnectorNode(failingStepConnector);
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
index 8560cb6f8bc..cba4b12f574 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorRepository.java
@@ -19,14 +19,24 @@ package org.apache.nifi.components.connector;
import org.apache.nifi.asset.Asset;
import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.bundle.BundleCoordinate;
+import org.apache.nifi.components.ConfigVerificationResult;
+import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.components.connector.components.FlowContextType;
+import org.apache.nifi.components.connector.secrets.SecretsManager;
import org.apache.nifi.controller.ParameterProviderNode;
import org.apache.nifi.controller.flow.FlowManager;
+import org.apache.nifi.controller.queue.QueueSize;
import org.apache.nifi.flow.Bundle;
import org.apache.nifi.flow.ScheduledState;
import org.apache.nifi.flow.VersionedConfigurationStep;
import org.apache.nifi.flow.VersionedConnector;
import org.apache.nifi.flow.VersionedConnectorValueReference;
+import org.apache.nifi.flow.VersionedExternalFlow;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.util.MockComponentLog;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@@ -35,7 +45,9 @@ import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -43,6 +55,7 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
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.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -50,6 +63,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
@@ -219,7 +233,7 @@ public class TestStandardConnectorRepository {
}
@Test
- public void testGetConnectorWithProviderOverridesWorkingConfig() {
+ public void testGetConnectorWithProviderOverridesWorkingConfig() throws FlowUpdateException {
final ConnectorConfigurationProvider provider = mock(ConnectorConfigurationProvider.class);
final StandardConnectorRepository repository = createRepositoryWithProvider(provider);
@@ -237,7 +251,7 @@ public class TestStandardConnectorRepository {
assertNotNull(result);
verify(connector).setName("External Name");
- verify(workingConfigContext).replaceProperties(eq("step1"), any(StepConfiguration.class));
+ verify(connector).replaceWorkingConfiguration(eq("step1"), any(StepConfiguration.class));
}
@Test
@@ -689,7 +703,7 @@ public class TestStandardConnectorRepository {
}
@Test
- public void testSyncFromProviderAppliesNifiUuidsDirectly() {
+ public void testSyncFromProviderAppliesNifiUuidsDirectly() throws FlowUpdateException {
final ConnectorConfigurationProvider provider = mock(ConnectorConfigurationProvider.class);
final StandardConnectorRepository repository = createRepositoryWithProvider(provider);
@@ -710,8 +724,85 @@ public class TestStandardConnectorRepository {
repository.getConnector("connector-1", ConnectorSyncMode.SYNC_WITH_PROVIDER);
- // Working config is updated with NiFi UUIDs as-is -- no translation in the repository
- verify(workingConfigContext).replaceProperties(eq("step1"), any(StepConfiguration.class));
+ // Working config is replaced with NiFi UUIDs as-is -- no translation in the repository
+ verify(connector).replaceWorkingConfiguration(eq("step1"), any(StepConfiguration.class));
+ }
+
+ @Test
+ public void testGetConnectorTriggersOnConfigurationStepConfiguredWhenValuesChange() throws FlowUpdateException {
+ final ConnectorConfigurationProvider provider = mock(ConnectorConfigurationProvider.class);
+ final StandardConnectorRepository repository = createRepositoryWithProvider(provider);
+
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createRealConnectorNode("connector-1", trackingConnector);
+ seedWorkingConfiguration(connectorNode, "step1", Map.of("prop1", "initial-value"));
+ repository.restoreConnector(connectorNode);
+
+ trackingConnector.reset();
+
+ final VersionedConfigurationStep externalStep = createVersionedStep("step1",
+ Map.of("prop1", createStringLiteralRef("updated-value")));
+ final ConnectorWorkingConfiguration externalConfig = new ConnectorWorkingConfiguration();
+ externalConfig.setName("connector-1");
+ externalConfig.setWorkingFlowConfiguration(List.of(externalStep));
+ when(provider.load("connector-1")).thenReturn(Optional.of(externalConfig));
+
+ final ConnectorNode result = repository.getConnector("connector-1", ConnectorSyncMode.SYNC_WITH_PROVIDER);
+
+ assertNotNull(result);
+ assertTrue(trackingConnector.wasOnStepConfiguredCalled("step1"));
+ }
+
+ @Test
+ public void testGetConnectorDoesNotTriggerOnConfigurationStepConfiguredWhenValuesUnchanged() throws FlowUpdateException {
+ final ConnectorConfigurationProvider provider = mock(ConnectorConfigurationProvider.class);
+ final StandardConnectorRepository repository = createRepositoryWithProvider(provider);
+
+ final TrackingConnector trackingConnector = new TrackingConnector();
+ final StandardConnectorNode connectorNode = createRealConnectorNode("connector-1", trackingConnector);
+ seedWorkingConfiguration(connectorNode, "step1", Map.of("prop1", "same-value"));
+ repository.restoreConnector(connectorNode);
+
+ trackingConnector.reset();
+
+ final VersionedConfigurationStep externalStep = createVersionedStep("step1",
+ Map.of("prop1", createStringLiteralRef("same-value")));
+ final ConnectorWorkingConfiguration externalConfig = new ConnectorWorkingConfiguration();
+ externalConfig.setName("connector-1");
+ externalConfig.setWorkingFlowConfiguration(List.of(externalStep));
+ when(provider.load("connector-1")).thenReturn(Optional.of(externalConfig));
+
+ final ConnectorNode result = repository.getConnector("connector-1", ConnectorSyncMode.SYNC_WITH_PROVIDER);
+
+ assertNotNull(result);
+ assertFalse(trackingConnector.wasOnStepConfiguredCalled("step1"));
+ }
+
+ @Test
+ public void testGetConnectorContinuesWhenOneStepFailsToReplace() throws FlowUpdateException {
+ final ConnectorConfigurationProvider provider = mock(ConnectorConfigurationProvider.class);
+ final StandardConnectorRepository repository = createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector = mock(ConnectorNode.class);
+ when(connector.getIdentifier()).thenReturn("connector-1");
+ doThrow(new FlowUpdateException("Simulated failure for step1"))
+ .when(connector).replaceWorkingConfiguration(eq("step1"), any(StepConfiguration.class));
+ repository.addConnector(connector);
+
+ final VersionedConfigurationStep stepOne = createVersionedStep("step1",
+ Map.of("prop1", createStringLiteralRef("value1")));
+ final VersionedConfigurationStep stepTwo = createVersionedStep("step2",
+ Map.of("prop2", createStringLiteralRef("value2")));
+ final ConnectorWorkingConfiguration externalConfig = new ConnectorWorkingConfiguration();
+ externalConfig.setName("connector-1");
+ externalConfig.setWorkingFlowConfiguration(List.of(stepOne, stepTwo));
+ when(provider.load("connector-1")).thenReturn(Optional.of(externalConfig));
+
+ final ConnectorNode result = repository.getConnector("connector-1", ConnectorSyncMode.SYNC_WITH_PROVIDER);
+
+ assertNotNull(result);
+ verify(connector).replaceWorkingConfiguration(eq("step1"), any(StepConfiguration.class));
+ verify(connector).replaceWorkingConfiguration(eq("step2"), any(StepConfiguration.class));
}
@Test
@@ -1379,6 +1470,80 @@ public class TestStandardConnectorRepository {
return vc;
}
+ private StandardConnectorNode createRealConnectorNode(final String identifier, final Connector connector) throws FlowUpdateException {
+ final ExtensionManager extensionManager = mock(ExtensionManager.class);
+ final AssetManager assetManager = mock(AssetManager.class);
+ final SecretsManager secretsManager = mock(SecretsManager.class);
+ when(secretsManager.getAllSecrets()).thenReturn(List.of());
+ when(secretsManager.getSecrets(anySet())).thenReturn(Collections.emptyMap());
+
+ final ProcessGroup managedProcessGroup = mock(ProcessGroup.class);
+ when(managedProcessGroup.purge()).thenReturn(CompletableFuture.completedFuture(null));
+ when(managedProcessGroup.getQueueSize()).thenReturn(new QueueSize(0, 0L));
+
+ final FlowContextFactory flowContextFactory = new FlowContextFactory() {
+ @Override
+ public FrameworkFlowContext createActiveFlowContext(final String connectorId, final ComponentLog connectorLogger, final Bundle bundle) {
+ final MutableConnectorConfigurationContext activeConfigurationContext =
+ new StandardConnectorConfigurationContext(assetManager, secretsManager);
+ return new StandardFlowContext(managedProcessGroup, activeConfigurationContext,
+ mock(ProcessGroupFacadeFactory.class), mock(ParameterContextFacadeFactory.class),
+ connectorLogger, FlowContextType.ACTIVE, bundle);
+ }
+
+ @Override
+ public FrameworkFlowContext createWorkingFlowContext(final String connectorId, final ComponentLog connectorLogger,
+ final MutableConnectorConfigurationContext currentConfiguration, final Bundle bundle) {
+ return new StandardFlowContext(managedProcessGroup, currentConfiguration,
+ mock(ProcessGroupFacadeFactory.class), mock(ParameterContextFacadeFactory.class),
+ connectorLogger, FlowContextType.WORKING, bundle);
+ }
+ };
+
+ final ConnectorStateTransition stateTransition = new StandardConnectorStateTransition("TestConnector-" + identifier);
+ final ConnectorValidationTrigger validationTrigger = new ConnectorValidationTrigger() {
+ @Override
+ public void triggerAsync(final ConnectorNode node) {
+ node.performValidation();
+ }
+
+ @Override
+ public void trigger(final ConnectorNode node) {
+ node.performValidation();
+ }
+ };
+
+ final ComponentLog componentLog = new MockComponentLog("TestConnector", connector);
+ final BundleCoordinate bundleCoordinate = new BundleCoordinate("org.apache.nifi", "test-standard-connector-node", "1.0.0");
+ final ConnectorDetails connectorDetails = new ConnectorDetails(connector, bundleCoordinate, componentLog);
+
+ final StandardConnectorNode node = new StandardConnectorNode(
+ identifier, mock(FlowManager.class), extensionManager, null, connectorDetails,
+ "TestConnector", connector.getClass().getCanonicalName(),
+ new StandardConnectorConfigurationContext(assetManager, secretsManager),
+ stateTransition, flowContextFactory, validationTrigger, false);
+
+ final FrameworkConnectorInitializationContext initializationContext = mock(FrameworkConnectorInitializationContext.class);
+ when(initializationContext.getSecretsManager()).thenReturn(secretsManager);
+ when(initializationContext.getAssetManager()).thenReturn(assetManager);
+
+ node.initializeConnector(initializationContext);
+ node.loadInitialFlow();
+ return node;
+ }
+
+ private void seedWorkingConfiguration(final StandardConnectorNode connectorNode, final String stepName,
+ final Map<String, String> properties) throws FlowUpdateException {
+ final Map<String, ConnectorValueReference> references = new HashMap<>();
+ for (final Map.Entry<String, String> entry : properties.entrySet()) {
+ references.put(entry.getKey(), new StringLiteralValue(entry.getValue()));
+ }
+ connectorNode.transitionStateForUpdating();
+ connectorNode.prepareForUpdate();
+ connectorNode.setConfiguration(stepName, new StepConfiguration(references));
+ connectorNode.applyUpdate();
+ }
+
private StandardConnectorRepository createRepositoryWithShortTimeout(final ConnectorConfigurationProvider provider) {
final StandardConnectorRepository repository = new StandardConnectorRepository();
final ConnectorRepositoryInitializationContext initContext = mock(ConnectorRepositoryInitializationContext.class);
@@ -1390,4 +1555,48 @@ public class TestStandardConnectorRepository {
repository.initialize(initContext);
return repository;
}
+
+ /**
+ * Test connector that tracks invocations of onStepConfigured so tests can assert which
+ * configuration steps were notified of changes.
+ */
+ private static class TrackingConnector extends AbstractConnector {
+ private final Set<String> onStepConfiguredCalls = new HashSet<>();
+
+ @Override
+ public VersionedExternalFlow getInitialFlow() {
+ return null;
+ }
+
+ @Override
+ public void prepareForUpdate(final FlowContext workingContext, final FlowContext activeContext) {
+ }
+
+ @Override
+ public List<ConfigurationStep> getConfigurationSteps() {
+ return List.of();
+ }
+
+ @Override
+ public void applyUpdate(final FlowContext workingContext, final FlowContext activeContext) {
+ }
+
+ @Override
+ protected void onStepConfigured(final String stepName, final FlowContext workingContext) {
+ onStepConfiguredCalls.add(stepName);
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verifyConfigurationStep(final String stepName, final Map<String, String> overrides, final FlowContext flowContext) {
+ return List.of();
+ }
+
+ public boolean wasOnStepConfiguredCalled(final String stepName) {
+ return onStepConfiguredCalls.contains(stepName);
+ }
+
+ public void reset() {
+ onStepConfiguredCalls.clear();
+ }
+ }
}