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 <[email protected]>
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 <[email protected]>
---
.../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();
+ }
+ }
}
