This is an automated email from the ASF dual-hosted git repository.
markap14 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 7fbc1b6a518 NIFI-15751 Improve Nifi Cluster Sync with Connectors
(#11079)
7fbc1b6a518 is described below
commit 7fbc1b6a51815526a1d36f7113716a87300e3eed
Author: Kevin Doran <[email protected]>
AuthorDate: Thu Apr 2 15:00:56 2026 -0400
NIFI-15751 Improve Nifi Cluster Sync with Connectors (#11079)
* NIFI-15751 Delegate connector sync from VFS to ConnectorRepository
Refactor VersionedFlowSynchronizer to delegate connector synchronization
entirely to ConnectorRepository, implementing state-aware behavior for all
ConnectorState values during flow inheritance.
- Introduce ConnectorSyncDirective (ALLOW/REJECT/REMOVE) returned by
ConnectorConfigurationProvider.verifySyncable, enabling providers to
override ScheduledState and supply working config + name
- Introduce ConnectorSyncResult enum for sync outcomes
- Move connector creation, config comparison, and lifecycle management
from VFS into StandardConnectorRepository.syncConnector
- Add configurable sync timeout (nifi.components.connectors.sync.timeout)
for waiting on transient ConnectorStates (STARTING, STOPPING, PURGING)
- Simplify VFS inheritConnectors to iterate and delegate
---
.../java/org/apache/nifi/util/NiFiProperties.java | 2 +
.../connector/ConnectorConfigurationProvider.java | 31 ++
.../connector/ConnectorSyncDirective.java | 144 +++++++
.../components/connector/ConnectorRepository.java | 23 ++
.../ConnectorRepositoryInitializationContext.java | 12 +
.../components/connector/ConnectorSyncResult.java | 119 ++++++
...StandardConnectorRepoInitializationContext.java | 12 +-
.../connector/StandardConnectorRepository.java | 342 ++++++++++++++++-
.../org/apache/nifi/controller/FlowController.java | 7 +-
.../serialization/VersionedFlowSynchronizer.java | 169 ++-------
.../connector/TestStandardConnectorRepository.java | 413 +++++++++++++++++++++
.../VersionedFlowSynchronizerTest.java | 127 ++++++-
.../system/connectors/ClusteredConnectorIT.java | 82 ++++
13 files changed, 1327 insertions(+), 156 deletions(-)
diff --git
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index eef06942f01..5b4648b23e8 100644
---
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -141,6 +141,8 @@ public class NiFiProperties extends ApplicationProperties {
// Connector Repository properties
public static final String CONNECTOR_REPOSITORY_IMPLEMENTATION =
"nifi.components.connectors.repository.implementation";
+ public static final String CONNECTOR_SYNC_TIMEOUT =
"nifi.components.connectors.sync.timeout";
+ public static final String DEFAULT_CONNECTOR_SYNC_TIMEOUT = "5 mins";
// Connector Configuration Provider properties
public static final String CONNECTOR_CONFIGURATION_PROVIDER_IMPLEMENTATION
= "nifi.components.connectors.configuration.provider.implementation";
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
index 4d2ff3ecc88..a0d5314a2f9 100644
---
a/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java
@@ -17,6 +17,8 @@
package org.apache.nifi.components.connector;
+import org.apache.nifi.flow.ScheduledState;
+
import java.io.InputStream;
import java.util.Optional;
@@ -99,6 +101,35 @@ public interface ConnectorConfigurationProvider {
*/
void verifyCreate(String connectorId);
+ /**
+ * Determines how the connector repository should handle synchronization
for the given
+ * connector during flow inheritance (cluster join). The provider examines
the external
+ * state of the connector and returns a {@link ConnectorSyncDirective}
indicating whether
+ * to allow, reject, or remove the connector.
+ *
+ * <p>When the directive action is {@link
ConnectorSyncDirective.Action#ALLOW}, the
+ * directive may optionally include:</p>
+ * <ul>
+ * <li>A {@link ConnectorWorkingConfiguration} with the provider's
working config and name
+ * (overriding the potentially stale values from the versioned
flow)</li>
+ * <li>A {@link ScheduledState} override (correcting stale run intent
from the versioned flow)</li>
+ * </ul>
+ *
+ * <p>This method combines the verify and load operations into a single
call to avoid
+ * redundant round-trips to the external store.</p>
+ *
+ * <p>The default implementation returns {@link
ConnectorSyncDirective#allow()} with no
+ * overrides, meaning the versioned flow's values are used for everything.
This is the
+ * behavior for Apache NiFi when no provider is configured.</p>
+ *
+ * @param connectorId the identifier of the connector to check
+ * @param proposedScheduledState the ScheduledState from the versioned flow
+ * @return a directive indicating how to handle this connector during sync
+ */
+ default ConnectorSyncDirective getSyncDirective(final String connectorId,
final ScheduledState proposedScheduledState) {
+ return ConnectorSyncDirective.allow();
+ }
+
/**
* Stores an asset to the local {@link org.apache.nifi.asset.AssetManager}
and to the external
* store. The provider records the NiFi UUID to external identifier
mapping in its local state.
diff --git
a/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncDirective.java
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncDirective.java
new file mode 100644
index 00000000000..3ca70c6ad11
--- /dev/null
+++
b/nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncDirective.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components.connector;
+
+import org.apache.nifi.flow.ScheduledState;
+
+/**
+ * Directive returned by {@link
ConnectorConfigurationProvider#getSyncDirective(String, ScheduledState)}
+ * indicating how the connector repository should handle synchronization for a
connector during
+ * flow inheritance.
+ */
+public class ConnectorSyncDirective {
+
+ /**
+ * The action the connector repository should take for this connector
during flow sync.
+ */
+ public enum Action {
+ /**
+ * Proceed with synchronization. The directive may optionally include a
+ * {@link ScheduledState} override and/or a {@link
ConnectorWorkingConfiguration}
+ * containing the provider's working config and name.
+ */
+ ALLOW,
+
+ /**
+ * Do not synchronize this connector. The connector should be created
locally
+ * (if not already present) and marked invalid so that a background
repair
+ * process can attempt synchronization later when conditions improve.
+ */
+ REJECT,
+
+ /**
+ * This connector should not exist on this node. If it exists locally,
remove
+ * it from the repository. If it does not exist, do not create it.
This is used
+ * when the external system indicates the connector is being deleted
or has been deleted.
+ */
+ REMOVE
+ }
+
+ private static final ConnectorSyncDirective ALLOW_DEFAULT = new
ConnectorSyncDirective(Action.ALLOW, null, null);
+ private static final ConnectorSyncDirective REJECT_DIRECTIVE = new
ConnectorSyncDirective(Action.REJECT, null, null);
+ private static final ConnectorSyncDirective REMOVE_DIRECTIVE = new
ConnectorSyncDirective(Action.REMOVE, null, null);
+
+ private final Action action;
+ private final ScheduledState scheduledStateOverride;
+ private final ConnectorWorkingConfiguration workingConfiguration;
+
+ private ConnectorSyncDirective(final Action action, final ScheduledState
scheduledStateOverride,
+ final ConnectorWorkingConfiguration
workingConfiguration) {
+ this.action = action;
+ this.scheduledStateOverride = scheduledStateOverride;
+ this.workingConfiguration = workingConfiguration;
+ }
+
+ /**
+ * Returns an ALLOW directive with no overrides. The connector repository
will use the
+ * versioned flow's name, working config, and ScheduledState as-is. This
is the default
+ * behavior when no {@link ConnectorConfigurationProvider} is configured
(Apache NiFi).
+ */
+ public static ConnectorSyncDirective allow() {
+ return ALLOW_DEFAULT;
+ }
+
+ /**
+ * Returns an ALLOW directive with the provider's working configuration
(name + working
+ * config steps) and no ScheduledState override.
+ *
+ * @param workingConfiguration the provider's working configuration
including name
+ */
+ public static ConnectorSyncDirective allow(final
ConnectorWorkingConfiguration workingConfiguration) {
+ return new ConnectorSyncDirective(Action.ALLOW, null,
workingConfiguration);
+ }
+
+ /**
+ * Returns an ALLOW directive with the provider's working configuration
and a
+ * ScheduledState override. The override replaces the versioned flow's
ScheduledState,
+ * which may be stale due to in-flight DPS tasks.
+ *
+ * @param workingConfiguration the provider's working configuration
including name
+ * @param scheduledStateOverride the ScheduledState to use instead of the
versioned flow's value
+ */
+ public static ConnectorSyncDirective allow(final
ConnectorWorkingConfiguration workingConfiguration,
+ final ScheduledState
scheduledStateOverride) {
+ return new ConnectorSyncDirective(Action.ALLOW,
scheduledStateOverride, workingConfiguration);
+ }
+
+ /**
+ * Returns a REJECT directive. The connector will be created locally (if
not already
+ * present) and marked invalid for background repair.
+ */
+ public static ConnectorSyncDirective reject() {
+ return REJECT_DIRECTIVE;
+ }
+
+ /**
+ * Returns a REMOVE directive. The connector should not exist on this node.
+ */
+ public static ConnectorSyncDirective remove() {
+ return REMOVE_DIRECTIVE;
+ }
+
+ public Action getAction() {
+ return action;
+ }
+
+ /**
+ * Returns the ScheduledState override, or {@code null} if the versioned
flow's
+ * ScheduledState should be used.
+ */
+ public ScheduledState getScheduledStateOverride() {
+ return scheduledStateOverride;
+ }
+
+ /**
+ * Returns the provider's working configuration (name + working config
steps),
+ * or {@code null} if the versioned flow's values should be used.
+ */
+ public ConnectorWorkingConfiguration getWorkingConfiguration() {
+ return workingConfiguration;
+ }
+
+ @Override
+ public String toString() {
+ return "ConnectorSyncDirective[action=" + action
+ + (scheduledStateOverride != null ? ",
scheduledStateOverride=" + scheduledStateOverride : "")
+ + (workingConfiguration != null ? ", hasWorkingConfig=true" :
"")
+ + "]";
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
index cfa5ae5f9a6..6c0a3ba43ea 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepository.java
@@ -21,6 +21,7 @@ import org.apache.nifi.asset.Asset;
import org.apache.nifi.components.connector.secrets.SecretsManager;
import org.apache.nifi.flow.Bundle;
import org.apache.nifi.flow.VersionedConfigurationStep;
+import org.apache.nifi.flow.VersionedConnector;
import java.io.IOException;
import java.io.InputStream;
@@ -59,6 +60,28 @@ public interface ConnectorRepository {
*/
void restoreConnector(ConnectorNode connector);
+ /**
+ * Synchronizes a connector during flow inheritance. This method handles
the full lifecycle:
+ * looking up or creating the local {@link ConnectorNode}, consulting the
+ * {@link ConnectorConfigurationProvider} for external state and
configuration, checking
+ * the current {@link ConnectorState}, waiting for transient states to
resolve, comparing
+ * configuration, applying changes via {@link #inheritConfiguration}, and
updating the
+ * run state.
+ *
+ * <p>The repository owns connector creation — if the connector does not
exist locally,
+ * the repository will create it (unless the provider directive indicates
the connector
+ * should be removed or rejected).</p>
+ *
+ * <p>When a {@link ConnectorConfigurationProvider} is configured, the
provider's
+ * {@link ConnectorConfigurationProvider#getSyncDirective(String,
org.apache.nifi.flow.ScheduledState)}
+ * method is called to obtain a {@link ConnectorSyncDirective} that may
override the
+ * working configuration, name, or ScheduledState from the versioned
flow.</p>
+ *
+ * @param versionedConnector the proposed connector from the versioned flow
+ * @return a {@link ConnectorSyncResult} indicating the outcome
+ */
+ ConnectorSyncResult syncConnector(VersionedConnector versionedConnector);
+
/**
* Removes the given Connector from the Repository
* @param connectorId the identifier of the Connector to remove
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
index 91bca1fcc95..7e6e1db372b 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java
@@ -23,6 +23,8 @@ import org.apache.nifi.controller.NodeTypeProvider;
import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.nar.ExtensionManager;
+import java.time.Duration;
+
public interface ConnectorRepositoryInitializationContext {
FlowManager getFlowManager();
@@ -47,4 +49,14 @@ public interface ConnectorRepositoryInitializationContext {
return null;
}
+ /**
+ * Returns the maximum time to wait for a connector in a transient state
(STARTING, STOPPING, PURGING)
+ * to reach a stable state during flow synchronization.
+ *
+ * @return the sync timeout duration, defaults to 5 minutes
+ */
+ default Duration getConnectorSyncTimeout() {
+ return Duration.ofMinutes(5);
+ }
+
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncResult.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncResult.java
new file mode 100644
index 00000000000..03a9a3f7413
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncResult.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components.connector;
+
+import org.apache.nifi.flow.ScheduledState;
+
+/**
+ * Result of a connector synchronization attempt during flow inheritance.
+ * Contains the outcome, the connector node (if applicable), and the effective
+ * ScheduledState that the caller should apply.
+ */
+public final class ConnectorSyncResult {
+
+ /**
+ * The outcome of the sync attempt.
+ */
+ public enum Outcome {
+ SYNCED,
+ SYNCED_CONFIG_UNCHANGED,
+ REJECTED,
+ FAILED,
+ REMOVED
+ }
+
+ private final Outcome outcome;
+ private final ConnectorNode connectorNode;
+ private final ScheduledState effectiveScheduledState;
+
+ private ConnectorSyncResult(final Outcome outcome, final ConnectorNode
connectorNode, final ScheduledState effectiveScheduledState) {
+ this.outcome = outcome;
+ this.connectorNode = connectorNode;
+ this.effectiveScheduledState = effectiveScheduledState;
+ }
+
+ /**
+ * Configuration was applied and the connector's run state should be
updated
+ * to match the effective ScheduledState.
+ */
+ public static ConnectorSyncResult synced(final ConnectorNode
connectorNode, final ScheduledState effectiveScheduledState) {
+ return new ConnectorSyncResult(Outcome.SYNCED, connectorNode,
effectiveScheduledState);
+ }
+
+ /**
+ * Connector configuration was unchanged relative to the proposed
versioned configuration;
+ * no configuration update was applied. The connector's run state should
still be updated
+ * if it differs from the effective ScheduledState.
+ */
+ public static ConnectorSyncResult syncedConfigUnchanged(final
ConnectorNode connectorNode, final ScheduledState effectiveScheduledState) {
+ return new ConnectorSyncResult(Outcome.SYNCED_CONFIG_UNCHANGED,
connectorNode, effectiveScheduledState);
+ }
+
+ /**
+ * The connector was in a state that prevents synchronization. The
connector
+ * has been marked invalid.
+ */
+ public static ConnectorSyncResult rejected(final ConnectorNode
connectorNode) {
+ return new ConnectorSyncResult(Outcome.REJECTED, connectorNode, null);
+ }
+
+ /**
+ * Synchronization was attempted but failed. The connector has been marked
invalid.
+ */
+ public static ConnectorSyncResult failed(final ConnectorNode
connectorNode) {
+ return new ConnectorSyncResult(Outcome.FAILED, connectorNode, null);
+ }
+
+ /**
+ * The connector was removed from this node because the external provider
+ * indicated it should not exist.
+ */
+ public static ConnectorSyncResult removed() {
+ return new ConnectorSyncResult(Outcome.REMOVED, null, null);
+ }
+
+ public Outcome getOutcome() {
+ return outcome;
+ }
+
+ /**
+ * Returns the connector node, or {@code null} if the connector was
removed.
+ */
+ public ConnectorNode getConnectorNode() {
+ return connectorNode;
+ }
+
+ /**
+ * Returns the effective ScheduledState that should be applied to the
connector,
+ * or {@code null} if the connector was rejected, failed, or removed.
+ * The caller is responsible for applying this state using the appropriate
+ * mechanism (e.g., {@code FlowController.startConnector()} which respects
+ * deferred-start semantics).
+ */
+ public ScheduledState getEffectiveScheduledState() {
+ return effectiveScheduledState;
+ }
+
+ @Override
+ public String toString() {
+ return "ConnectorSyncResult[outcome=" + outcome
+ + (connectorNode != null ? ", connector=" +
connectorNode.getIdentifier() : "")
+ + (effectiveScheduledState != null ? ", effectiveState=" +
effectiveScheduledState : "")
+ + "]";
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
index a0661fa58f9..10fe9b49af0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepoInitializationContext.java
@@ -23,6 +23,8 @@ import org.apache.nifi.controller.NodeTypeProvider;
import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.nar.ExtensionManager;
+import java.time.Duration;
+
public class StandardConnectorRepoInitializationContext implements
ConnectorRepositoryInitializationContext {
private final FlowManager flowManager;
private final ExtensionManager extensionManager;
@@ -31,6 +33,7 @@ public class StandardConnectorRepoInitializationContext
implements ConnectorRepo
private final NodeTypeProvider nodeTypeProvider;
private final ConnectorRequestReplicator requestReplicator;
private final ConnectorConfigurationProvider
connectorConfigurationProvider;
+ private final Duration connectorSyncTimeout;
public StandardConnectorRepoInitializationContext(final FlowManager
flowManager,
final ExtensionManager
extensionManager,
@@ -38,7 +41,8 @@ public class StandardConnectorRepoInitializationContext
implements ConnectorRepo
final AssetManager
assetManager,
final NodeTypeProvider
nodeTypeProvider,
final
ConnectorRequestReplicator requestReplicator,
- final
ConnectorConfigurationProvider connectorConfigurationProvider) {
+ final
ConnectorConfigurationProvider connectorConfigurationProvider,
+ final Duration
connectorSyncTimeout) {
this.flowManager = flowManager;
this.extensionManager = extensionManager;
this.secretsManager = secretsManager;
@@ -46,6 +50,7 @@ public class StandardConnectorRepoInitializationContext
implements ConnectorRepo
this.nodeTypeProvider = nodeTypeProvider;
this.requestReplicator = requestReplicator;
this.connectorConfigurationProvider = connectorConfigurationProvider;
+ this.connectorSyncTimeout = connectorSyncTimeout;
}
@Override
@@ -82,4 +87,9 @@ public class StandardConnectorRepoInitializationContext
implements ConnectorRepo
public ConnectorConfigurationProvider getConnectorConfigurationProvider() {
return connectorConfigurationProvider;
}
+
+ @Override
+ public Duration getConnectorSyncTimeout() {
+ return connectorSyncTimeout;
+ }
}
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 2df018c6382..0b0382ccfc2 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
@@ -20,14 +20,20 @@ package org.apache.nifi.components.connector;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.asset.Asset;
import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.components.connector.secrets.SecretsManager;
+import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.engine.FlowEngine;
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.nar.ExtensionManager;
import org.apache.nifi.nar.NarCloseable;
+import org.apache.nifi.util.BundleUtils;
import org.apache.nifi.util.ReflectionUtils;
+import org.apache.nifi.web.api.dto.BundleDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,34 +45,43 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import java.util.function.Function;
+import java.util.stream.Collectors;
public class StandardConnectorRepository implements ConnectorRepository {
private static final Logger logger =
LoggerFactory.getLogger(StandardConnectorRepository.class);
+ private static final Duration SYNC_POLL_INTERVAL = Duration.ofSeconds(2);
private final Map<String, ConnectorNode> connectors = new
ConcurrentHashMap<>();
private final FlowEngine lifecycleExecutor = new FlowEngine(8, "NiFi
Connector Lifecycle");
+ private volatile FlowManager flowManager;
private volatile ExtensionManager extensionManager;
private volatile ConnectorRequestReplicator requestReplicator;
private volatile SecretsManager secretsManager;
private volatile AssetManager assetManager;
private volatile ConnectorConfigurationProvider configurationProvider;
+ private volatile Duration syncTimeout;
@Override
public void initialize(final ConnectorRepositoryInitializationContext
context) {
logger.debug("Initializing ConnectorRepository");
+ this.flowManager = context.getFlowManager();
this.extensionManager = context.getExtensionManager();
this.requestReplicator = context.getRequestReplicator();
this.secretsManager = context.getSecretsManager();
this.assetManager = context.getAssetManager();
this.configurationProvider =
context.getConnectorConfigurationProvider();
- logger.debug("Successfully initialized ConnectorRepository with
configurationProvider={}", configurationProvider != null ?
configurationProvider.getClass().getSimpleName() : "null");
+ this.syncTimeout = context.getConnectorSyncTimeout();
+ logger.debug("Successfully initialized ConnectorRepository with
configurationProvider={}, syncTimeout={}",
+ configurationProvider != null ?
configurationProvider.getClass().getSimpleName() : "null", syncTimeout);
}
@Override
@@ -91,16 +106,298 @@ public class StandardConnectorRepository implements
ConnectorRepository {
logger.debug("Successfully restored {}", connector);
}
+ @Override
+ public ConnectorSyncResult syncConnector(final VersionedConnector
versionedConnector) {
+ final String connectorId = versionedConnector.getInstanceIdentifier();
+ final ScheduledState proposedScheduledState =
versionedConnector.getScheduledState();
+ logger.debug("syncConnector called for connector [{}]", connectorId);
+
+ // Consult the provider for external state checks and working config
+ final ConnectorSyncDirective directive;
+ if (configurationProvider != null) {
+ try {
+ directive =
configurationProvider.getSyncDirective(connectorId, proposedScheduledState);
+ } catch (final Exception e) {
+ logger.error("Configuration provider threw exception during
getSyncDirective for connector [{}]: {}", connectorId, e.getMessage(), e);
+ final ConnectorNode existingNode =
ensureConnectorNodeExists(versionedConnector);
+ existingNode.markInvalid("Flow Synchronization Failure",
+ "Configuration provider error during synchronization:
" + e.getMessage());
+ return ConnectorSyncResult.failed(existingNode);
+ }
+ } else {
+ directive = ConnectorSyncDirective.allow();
+ }
+
+ logger.debug("Connector [{}] sync directive: {}", connectorId,
directive);
+
+ // Handle REMOVE: connector should not exist on this node
+ if (directive.getAction() == ConnectorSyncDirective.Action.REMOVE) {
+ final ConnectorNode existingNode = connectors.get(connectorId);
+ if (existingNode != null) {
+ logger.info("Removing connector [{}] (state={}) from local
repository per provider REMOVE directive",
+ connectorId, existingNode.getCurrentState());
+ stopConnectorAndAwait(existingNode);
+ purgeConnectorAndAwait(existingNode);
+ logger.debug("Connector [{}] stopped and purged (state={});
calling removeConnector", connectorId, existingNode.getCurrentState());
+ removeConnector(connectorId);
+ logger.info("Successfully removed connector [{}] per REMOVE
directive", connectorId);
+ } else {
+ logger.debug("Connector [{}] not present locally; REMOVE
directive is a no-op", connectorId);
+ }
+ return ConnectorSyncResult.removed();
+ }
+
+ // Handle REJECT: create if needed (for background repair), mark
invalid
+ if (directive.getAction() == ConnectorSyncDirective.Action.REJECT) {
+ final ConnectorNode node =
ensureConnectorNodeExists(versionedConnector);
+ logger.warn("Connector [{}] rejected by provider during sync;
marking invalid", connectorId);
+ node.markInvalid("Flow Synchronization Failure",
+ "Configuration provider rejected synchronization for this
connector");
+ return ConnectorSyncResult.rejected(node);
+ }
+
+ // ALLOW: proceed with sync
+ // Look up or create the connector node
+ ConnectorNode connector = connectors.get(connectorId);
+ final boolean isNewConnector = connector == null;
+ if (isNewConnector) {
+ connector = createConnectorNode(versionedConnector);
+ connectors.put(connectorId, connector);
+ logger.info("Created new connector node [{}] of type [{}]",
connectorId, versionedConnector.getType());
+ }
+
+ // Check local ConnectorState and handle per the agreed behavior matrix
+ ConnectorState currentState = connector.getCurrentState();
+
+ if (isWaitableState(currentState)) {
+ logger.info("{} is in transient state {}; waiting for stable state
(timeout={})", connector, currentState, syncTimeout);
+ currentState = waitForStableState(connector, currentState);
+ if (isWaitableState(currentState)) {
+ logger.error("{} timed out waiting for transient state {} to
resolve", connector, currentState);
+ connector.markInvalid("Flow Synchronization Failure",
+ "Timed out waiting for connector to leave transient
state " + currentState);
+ return ConnectorSyncResult.rejected(connector);
+ }
+ }
+
+ if (isRejectableState(currentState)) {
+ logger.warn("{} is in state {} which cannot be synchronized;
marking invalid", connector, currentState);
+ connector.markInvalid("Flow Synchronization Failure",
+ "Connector cannot be synchronized while in state " +
currentState);
+ return ConnectorSyncResult.rejected(connector);
+ }
+
+ // Determine effective name, working config, and ScheduledState
+ final ConnectorWorkingConfiguration providerConfig =
directive.getWorkingConfiguration();
+
+ final String effectiveName = (providerConfig != null &&
providerConfig.getName() != null)
+ ? providerConfig.getName()
+ : versionedConnector.getName();
+
+ final List<VersionedConfigurationStep> effectiveWorkingConfig =
(providerConfig != null && providerConfig.getWorkingFlowConfiguration() != null)
+ ? providerConfig.getWorkingFlowConfiguration()
+ : versionedConnector.getWorkingFlowConfiguration();
+
+ final List<VersionedConfigurationStep> effectiveActiveConfig =
versionedConnector.getActiveFlowConfiguration();
+
+ final ScheduledState effectiveScheduledState =
(directive.getScheduledStateOverride() != null)
+ ? directive.getScheduledStateOverride()
+ : proposedScheduledState;
+
+ // Set name locally (no provider.save())
+ connector.setName(effectiveName);
+
+ // Compare config and sync if changed
+ final boolean wasRunning = currentState == ConnectorState.RUNNING;
+ final boolean configChanged = isNewConnector ||
isConfigurationUpdated(connector, effectiveActiveConfig,
effectiveWorkingConfig);
+
+ if (configChanged) {
+ logger.info("{} configuration needs synchronization", connector);
+ try {
+ inheritConfiguration(connector, effectiveActiveConfig,
effectiveWorkingConfig, versionedConnector.getBundle());
+ } catch (final Exception e) {
+ logger.error("{} failed to inherit configuration", connector,
e);
+ if (wasRunning) {
+ try {
+ stopConnector(connector);
+ } catch (final Exception stopEx) {
+ logger.error("{} also failed to stop after sync
failure", connector, stopEx);
+ }
+ }
+ return ConnectorSyncResult.failed(connector);
+ }
+ } else {
+ logger.debug("{} configuration is up to date, no update
necessary", connector);
+ }
+
+ return configChanged
+ ? ConnectorSyncResult.synced(connector,
effectiveScheduledState)
+ : ConnectorSyncResult.syncedConfigUnchanged(connector,
effectiveScheduledState);
+ }
+
+ private ConnectorNode ensureConnectorNodeExists(final VersionedConnector
versionedConnector) {
+ final String connectorId = versionedConnector.getInstanceIdentifier();
+ ConnectorNode node = connectors.get(connectorId);
+ if (node == null) {
+ node = createConnectorNode(versionedConnector);
+ connectors.put(connectorId, node);
+ logger.info("Created connector node [{}] for reject/error
handling", connectorId);
+ }
+ return node;
+ }
+
+ private ConnectorNode createConnectorNode(final VersionedConnector
versionedConnector) {
+ final BundleCoordinate coordinate =
resolveBundleCoordinate(versionedConnector.getBundle(),
versionedConnector.getType());
+ return flowManager.createConnector(versionedConnector.getType(),
versionedConnector.getInstanceIdentifier(), coordinate, false, true);
+ }
+
+ private BundleCoordinate resolveBundleCoordinate(final Bundle bundle,
final String componentType) {
+ try {
+ final BundleDTO bundleDto = new BundleDTO(bundle.getGroup(),
bundle.getArtifact(), bundle.getVersion());
+ return BundleUtils.getCompatibleBundle(extensionManager,
componentType, bundleDto);
+ } catch (final IllegalStateException e) {
+ return new BundleCoordinate(bundle.getGroup(),
bundle.getArtifact(), bundle.getVersion());
+ }
+ }
+
+ private boolean isWaitableState(final ConnectorState state) {
+ return state == ConnectorState.STARTING
+ || state == ConnectorState.STOPPING
+ || state == ConnectorState.PURGING;
+ }
+
+ private boolean isRejectableState(final ConnectorState state) {
+ return state == ConnectorState.DRAINING
+ || state == ConnectorState.PREPARING_FOR_UPDATE
+ || state == ConnectorState.UPDATING;
+ }
+
+ private ConnectorState waitForStableState(final ConnectorNode connector,
final ConnectorState initialState) {
+ final long deadlineNanos = System.nanoTime() + syncTimeout.toNanos();
+ int iterations = 0;
+ ConnectorState current = initialState;
+
+ while (isWaitableState(current) && System.nanoTime() < deadlineNanos) {
+ try {
+ Thread.sleep(SYNC_POLL_INTERVAL.toMillis());
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("{} interrupted while waiting for stable state",
connector);
+ return current;
+ }
+
+ current = connector.getCurrentState();
+ iterations++;
+ if (iterations % 10 == 0) {
+ final long elapsedSeconds = Duration.ofNanos(System.nanoTime()
- (deadlineNanos - syncTimeout.toNanos())).toSeconds();
+ logger.info("Still waiting for {} to leave state {}; current
state is {}; elapsed={}s",
+ connector, initialState, current, elapsedSeconds);
+ }
+ }
+
+ return current;
+ }
+
+ // --- Configuration comparison ---
+
+ private boolean isConfigurationUpdated(final ConnectorNode
existingConnector,
+ final
List<VersionedConfigurationStep> effectiveActiveConfig,
+ final
List<VersionedConfigurationStep> effectiveWorkingConfig) {
+ final ConnectorConfiguration activeConfig =
existingConnector.getActiveFlowContext().getConfigurationContext().toConnectorConfiguration();
+ final boolean activeContextChanged =
isConfigurationUpdated(activeConfig, effectiveActiveConfig);
+
+ final ConnectorConfiguration workingConfig =
existingConnector.getWorkingFlowContext().getConfigurationContext().toConnectorConfiguration();
+ final boolean workingContextChanged =
isConfigurationUpdated(workingConfig, effectiveWorkingConfig);
+
+ return activeContextChanged || workingContextChanged;
+ }
+
+ private boolean isConfigurationUpdated(final ConnectorConfiguration
existingConfiguration, final List<VersionedConfigurationStep>
versionedConfigurationSteps) {
+ if (versionedConfigurationSteps == null ||
versionedConfigurationSteps.isEmpty()) {
+ return existingConfiguration != null &&
!existingConfiguration.getNamedStepConfigurations().isEmpty();
+ }
+
+ final Set<NamedStepConfiguration> existingStepConfigurations =
existingConfiguration.getNamedStepConfigurations();
+ if (existingStepConfigurations.size() !=
versionedConfigurationSteps.size()) {
+ return true;
+ }
+
+ final Map<String, NamedStepConfiguration> existingStepsByName =
existingStepConfigurations.stream()
+ .collect(Collectors.toMap(NamedStepConfiguration::stepName,
Function.identity()));
+
+ for (final VersionedConfigurationStep versionedStep :
versionedConfigurationSteps) {
+ final NamedStepConfiguration existingStep =
existingStepsByName.get(versionedStep.getName());
+ if (existingStep == null) {
+ return true;
+ }
+
+ if (isConfigurationStepUpdated(existingStep, versionedStep)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private boolean isConfigurationStepUpdated(final NamedStepConfiguration
existingStep, final VersionedConfigurationStep versionedStep) {
+ final Map<String, ConnectorValueReference> existingProperties =
existingStep.configuration().getPropertyValues();
+
+ final Map<String, VersionedConnectorValueReference>
versionedProperties = versionedStep.getProperties();
+ if (versionedProperties == null || versionedProperties.isEmpty()) {
+ return existingProperties != null && !existingProperties.isEmpty();
+ }
+
+ if (existingProperties == null || existingProperties.size() !=
versionedProperties.size()) {
+ return true;
+ }
+
+ for (final Map.Entry<String, VersionedConnectorValueReference>
versionedEntry : versionedProperties.entrySet()) {
+ final String propertyName = versionedEntry.getKey();
+ final VersionedConnectorValueReference versionedRef =
versionedEntry.getValue();
+ final ConnectorValueReference existingRef =
existingProperties.get(propertyName);
+
+ if (!valueReferencesEqual(versionedRef, existingRef)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private boolean valueReferencesEqual(final
VersionedConnectorValueReference versionedReference, final
ConnectorValueReference existingReference) {
+ if (versionedReference == null && existingReference == null) {
+ return true;
+ }
+ if (versionedReference == null || existingReference == null) {
+ return false;
+ }
+
+ final String versionedValueType = versionedReference.getValueType();
+ final String existingValueType =
existingReference.getValueType().name();
+ if (!Objects.equals(versionedValueType, existingValueType)) {
+ return false;
+ }
+
+ return switch (existingReference) {
+ case StringLiteralValue stringLiteral ->
Objects.equals(stringLiteral.getValue(), versionedReference.getValue());
+ case AssetReference assetRef ->
Objects.equals(assetRef.getAssetIdentifiers(),
versionedReference.getAssetIds());
+ case SecretReference secretRef ->
Objects.equals(secretRef.getProviderId(), versionedReference.getProviderId())
+ && Objects.equals(secretRef.getSecretName(),
versionedReference.getSecretName());
+ };
+ }
+
@Override
public void removeConnector(final String connectorId) {
- logger.debug("Removing {}", connectorId);
+ logger.debug("Removing connector [{}]", connectorId);
final ConnectorNode connectorNode = connectors.get(connectorId);
if (connectorNode == null) {
throw new IllegalStateException("No connector found with ID " +
connectorId);
}
+ logger.debug("Verifying connector [{}] (state={}) can be deleted",
connectorId, connectorNode.getCurrentState());
connectorNode.verifyCanDelete();
if (configurationProvider != null) {
+ logger.debug("Notifying configuration provider of connector [{}]
deletion", connectorId);
configurationProvider.delete(connectorId);
}
connectors.remove(connectorId);
@@ -111,6 +408,47 @@ public class StandardConnectorRepository implements
ConnectorRepository {
}
extensionManager.removeInstanceClassLoader(connectorId);
+ logger.info("Connector [{}] removed from repository", connectorId);
+ }
+
+ private void stopConnectorAndAwait(final ConnectorNode connector) {
+ final String connectorId = connector.getIdentifier();
+ final ConnectorState currentState = connector.getCurrentState();
+ if (currentState == ConnectorState.STOPPED || currentState ==
ConnectorState.UPDATE_FAILED || currentState == ConnectorState.UPDATED) {
+ logger.debug("Connector [{}] is already in state {}; no stop
needed", connectorId, currentState);
+ return;
+ }
+
+ logger.info("Stopping connector [{}] (current state: {}) and awaiting
completion", connectorId, currentState);
+ try {
+ final Future<Void> stopFuture = stopConnector(connector);
+ stopFuture.get(syncTimeout.toMillis(),
java.util.concurrent.TimeUnit.MILLISECONDS);
+ logger.debug("Connector [{}] stopped successfully", connectorId);
+ } catch (final java.util.concurrent.TimeoutException e) {
+ logger.warn("Timed out waiting for connector [{}] to stop",
connectorId);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Interrupted while waiting for connector [{}] to
stop", connectorId);
+ } catch (final Exception e) {
+ logger.warn("Failed to stop connector [{}]: {}", connectorId,
e.getMessage(), e);
+ }
+ }
+
+ private void purgeConnectorAndAwait(final ConnectorNode connector) {
+ final String connectorId = connector.getIdentifier();
+ try {
+ logger.debug("Purging FlowFiles for connector [{}] before
removal", connectorId);
+ final Future<Void> purgeFuture = connector.purgeFlowFiles("Flow
Synchronization");
+ purgeFuture.get(syncTimeout.toMillis(),
java.util.concurrent.TimeUnit.MILLISECONDS);
+ logger.debug("Connector [{}] purged successfully", connectorId);
+ } catch (final java.util.concurrent.TimeoutException e) {
+ logger.warn("Timed out waiting for connector [{}] to purge
FlowFiles", connectorId);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Interrupted while waiting for connector [{}] to purge
FlowFiles", connectorId);
+ } catch (final Exception e) {
+ logger.warn("Failed to purge FlowFiles for connector [{}]: {}",
connectorId, e.getMessage(), e);
+ }
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index c3de110791e..26137a6ae53 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -934,6 +934,10 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
extensionManager.discoverExtensions(extensionManager.getAllBundles(),
additionalExtensionTypes, false);
final ConnectorRepository created =
NarThreadContextClassLoader.createInstance(extensionManager,
implementationClassName, ConnectorRepository.class, properties);
+ final String syncTimeoutValue =
properties.getProperty(NiFiProperties.CONNECTOR_SYNC_TIMEOUT,
NiFiProperties.DEFAULT_CONNECTOR_SYNC_TIMEOUT);
+ final long syncTimeoutMillis = (long)
FormatUtils.getTimeDuration(syncTimeoutValue, TimeUnit.MILLISECONDS);
+ final Duration connectorSyncTimeout =
Duration.ofMillis(syncTimeoutMillis);
+
final ConnectorRepositoryInitializationContext
initializationContext = new StandardConnectorRepoInitializationContext(
flowManager,
extensionManager,
@@ -941,7 +945,8 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
assetManager,
nodeTypeProvider,
requestReplicator,
- connectorConfigurationProvider
+ connectorConfigurationProvider,
+ connectorSyncTimeout
);
synchronized (created) {
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 3e002d5450e..2a07ac621fe 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
@@ -26,15 +26,9 @@ import org.apache.nifi.authorization.ManagedAuthorizer;
import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.cluster.protocol.StandardDataFlow;
-import org.apache.nifi.components.connector.AssetReference;
-import org.apache.nifi.components.connector.ConnectorConfiguration;
import org.apache.nifi.components.connector.ConnectorNode;
import org.apache.nifi.components.connector.ConnectorRepository;
-import org.apache.nifi.components.connector.ConnectorValueReference;
-import org.apache.nifi.components.connector.FlowUpdateException;
-import org.apache.nifi.components.connector.NamedStepConfiguration;
-import org.apache.nifi.components.connector.SecretReference;
-import org.apache.nifi.components.connector.StringLiteralValue;
+import org.apache.nifi.components.connector.ConnectorSyncResult;
import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.Position;
@@ -64,9 +58,7 @@ import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.flow.ScheduledState;
import org.apache.nifi.flow.VersionedAsset;
import org.apache.nifi.flow.VersionedComponent;
-import org.apache.nifi.flow.VersionedConfigurationStep;
import org.apache.nifi.flow.VersionedConnector;
-import org.apache.nifi.flow.VersionedConnectorValueReference;
import org.apache.nifi.flow.VersionedControllerService;
import org.apache.nifi.flow.VersionedExternalFlow;
import org.apache.nifi.flow.VersionedFlowAnalysisRule;
@@ -1042,9 +1034,6 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
}
private void inheritConnectors(final FlowController flowController, final
VersionedDataflow dataflow) {
- // TODO: We need to delete any Connectors that are no longer part of
the flow.
- // This means we need to drain the Connector first, then stop
it, then delete it. If unable to drain, we must fail...
- // perhaps we need a DRAINING state? Or do we just delete
it and drop the data?
final ConnectorRepository connectorRepository =
flowController.getConnectorRepository();
final Set<String> proposedConnectorIds = new HashSet<>();
@@ -1052,145 +1041,39 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
for (final VersionedConnector versionedConnector :
dataflow.getConnectors()) {
proposedConnectorIds.add(versionedConnector.getInstanceIdentifier());
- final ConnectorNode existingConnector =
connectorRepository.getConnector(versionedConnector.getInstanceIdentifier());
- if (existingConnector == null) {
- logger.info("Connector {} of type {} with name {} is not
in the current flow. Will add Connector.",
- versionedConnector.getInstanceIdentifier(),
versionedConnector.getType(), versionedConnector.getName());
+ final ConnectorSyncResult result =
connectorRepository.syncConnector(versionedConnector);
+ logger.info("Connector [{}] sync result: {}",
versionedConnector.getInstanceIdentifier(), result);
- addConnector(versionedConnector, flowController);
- } else if (isConnectorConfigurationUpdated(existingConnector,
versionedConnector)) {
- logger.info("{} configuration has changed, updating
configuration", existingConnector);
- updateConnector(versionedConnector, flowController);
- } else {
- logger.debug("{} configuration is up to date, no update
necessary", existingConnector);
+ if (result.getEffectiveScheduledState() != null &&
result.getConnectorNode() != null) {
+ if (result.getEffectiveScheduledState() ==
ScheduledState.RUNNING) {
+
flowController.startConnector(result.getConnectorNode());
+ } else if (result.getEffectiveScheduledState() ==
ScheduledState.ENABLED) {
+
connectorRepository.stopConnector(result.getConnectorNode());
+ }
}
}
}
for (final ConnectorNode existingConnector :
connectorRepository.getConnectors()) {
if
(!proposedConnectorIds.contains(existingConnector.getIdentifier())) {
- logger.info("Connector {} is no longer part of the proposed
flow. Will remove Connector.", existingConnector);
- connectorRepository.stopConnector(existingConnector);
-
connectorRepository.removeConnector(existingConnector.getIdentifier());
- }
- }
- }
-
- private boolean isConnectorConfigurationUpdated(final ConnectorNode
existingConnector, final VersionedConnector versionedConnector) {
- final ConnectorConfiguration activeConfig =
existingConnector.getActiveFlowContext().getConfigurationContext().toConnectorConfiguration();
- final List<VersionedConfigurationStep> versionedActiveConfig =
versionedConnector.getActiveFlowConfiguration();
- final boolean activeContextChanged =
isConnectorConfigurationUpdated(activeConfig, versionedActiveConfig);
-
- final ConnectorConfiguration workingConfig =
existingConnector.getWorkingFlowContext().getConfigurationContext().toConnectorConfiguration();
- final List<VersionedConfigurationStep> versionedWorkingConfig =
versionedConnector.getWorkingFlowConfiguration();
- final boolean workingContextChanged =
isConnectorConfigurationUpdated(workingConfig, versionedWorkingConfig);
-
- return activeContextChanged || workingContextChanged;
- }
-
- private boolean isConnectorConfigurationUpdated(final
ConnectorConfiguration existingConfiguration, final
List<VersionedConfigurationStep> versionedConfigurationSteps) {
- if (versionedConfigurationSteps == null ||
versionedConfigurationSteps.isEmpty()) {
- return existingConfiguration != null &&
!existingConfiguration.getNamedStepConfigurations().isEmpty();
- }
-
- final Set<NamedStepConfiguration> existingStepConfigurations =
existingConfiguration.getNamedStepConfigurations();
- if (existingStepConfigurations.size() !=
versionedConfigurationSteps.size()) {
- return true;
- }
-
- final Map<String, NamedStepConfiguration> existingStepsByName =
existingStepConfigurations.stream()
- .collect(Collectors.toMap(NamedStepConfiguration::stepName,
Function.identity()));
-
- for (final VersionedConfigurationStep versionedStep :
versionedConfigurationSteps) {
- final NamedStepConfiguration existingStep =
existingStepsByName.get(versionedStep.getName());
- if (existingStep == null) {
- return true;
- }
-
- if (isConfigurationStepUpdated(existingStep, versionedStep)) {
- return true;
- }
- }
-
- return false;
- }
-
- private boolean isConfigurationStepUpdated(final NamedStepConfiguration
existingStep, final VersionedConfigurationStep versionedStep) {
- final Map<String, ConnectorValueReference> existingProperties =
existingStep.configuration().getPropertyValues();
-
- final Map<String, VersionedConnectorValueReference>
versionedProperties = versionedStep.getProperties();
- if (versionedProperties == null || versionedProperties.isEmpty()) {
- return existingProperties != null && !existingProperties.isEmpty();
- }
-
- if (existingProperties == null || existingProperties.size() !=
versionedProperties.size()) {
- return true;
- }
-
- for (final Map.Entry<String, VersionedConnectorValueReference>
versionedEntry : versionedProperties.entrySet()) {
- final String propertyName = versionedEntry.getKey();
- final VersionedConnectorValueReference versionedRef =
versionedEntry.getValue();
- final ConnectorValueReference existingRef =
existingProperties.get(propertyName);
-
- if (!equals(versionedRef, existingRef)) {
- return true;
- }
- }
-
- return false;
- }
-
- private boolean equals(final VersionedConnectorValueReference
versionedReference, final ConnectorValueReference existingReference) {
- if (versionedReference == null && existingReference == null) {
- return true;
- }
- if (versionedReference == null || existingReference == null) {
- return false;
- }
-
- final String versionedValueType = versionedReference.getValueType();
- final String existingValueType =
existingReference.getValueType().name();
- if (!Objects.equals(versionedValueType, existingValueType)) {
- return false;
- }
-
- return switch (existingReference) {
- case StringLiteralValue stringLiteral ->
Objects.equals(stringLiteral.getValue(), versionedReference.getValue());
- case AssetReference assetRef ->
Objects.equals(assetRef.getAssetIdentifiers(),
versionedReference.getAssetIds());
- case SecretReference secretRef ->
Objects.equals(secretRef.getProviderId(), versionedReference.getProviderId())
- &&
Objects.equals(secretRef.getSecretName(), versionedReference.getSecretName());
- };
- }
-
- private void addConnector(final VersionedConnector versionedConnector,
final FlowController flowController) {
- final ConnectorRepository connectorRepository =
flowController.getConnectorRepository();
- final FlowManager flowManager = flowController.getFlowManager();
- final BundleCoordinate coordinate =
createBundleCoordinate(extensionManager, versionedConnector.getBundle(),
versionedConnector.getType());
- final ConnectorNode connectorNode =
flowManager.createConnector(versionedConnector.getType(),
versionedConnector.getInstanceIdentifier(), coordinate, false, true);
- connectorRepository.restoreConnector(connectorNode);
- updateConnector(versionedConnector, flowController);
- }
-
-
- private void updateConnector(final VersionedConnector versionedConnector,
final FlowController flowController) {
- final ConnectorRepository connectorRepository =
flowController.getConnectorRepository();
- final ConnectorNode connectorNode =
connectorRepository.getConnector(versionedConnector.getInstanceIdentifier());
-
- connectorRepository.updateConnector(connectorNode,
versionedConnector.getName());
-
- try {
- final List<VersionedConfigurationStep> activeFlowConfig =
versionedConnector.getActiveFlowConfiguration();
- final List<VersionedConfigurationStep> workingFlowConfig =
versionedConnector.getWorkingFlowConfiguration();
- connectorRepository.inheritConfiguration(connectorNode,
activeFlowConfig, workingFlowConfig, versionedConnector.getBundle());
-
- final ScheduledState desiredState =
versionedConnector.getScheduledState();
- if (desiredState == ScheduledState.RUNNING) {
- flowController.startConnector(connectorNode);
- } else if (desiredState == ScheduledState.ENABLED) {
- connectorRepository.stopConnector(connectorNode);
+ logger.info("Connector [{}] (state={}) is no longer part of
the proposed flow. Stopping and removing.",
+ existingConnector.getIdentifier(),
existingConnector.getCurrentState());
+ try {
+ logger.debug("Stopping orphan connector [{}] before
removal", existingConnector.getIdentifier());
+ connectorRepository.stopConnector(existingConnector).get();
+ logger.debug("Orphan connector [{}] stopped (state={});
purging data before removal",
+ existingConnector.getIdentifier(),
existingConnector.getCurrentState());
+ existingConnector.purgeFlowFiles("Flow
Synchronization").get();
+ logger.debug("Orphan connector [{}] purged; proceeding
with removal", existingConnector.getIdentifier());
+
connectorRepository.removeConnector(existingConnector.getIdentifier());
+ logger.info("Successfully removed orphan connector [{}]",
existingConnector.getIdentifier());
+ } catch (final Exception e) {
+ logger.error("Failed to remove Connector [{}] during flow
inheritance. Connector will be marked invalid.",
+ existingConnector.getIdentifier(), e);
+ existingConnector.markInvalid("Flow Synchronization
Failure",
+ "Connector could not be removed during flow sync:
" + e.getMessage());
+ }
}
- } catch (final FlowUpdateException e) {
- logger.error("{} failed to inherit configuration", connectorNode,
e);
}
}
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 4089b274819..f6e50200852 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,7 +19,11 @@ package org.apache.nifi.components.connector;
import org.apache.nifi.asset.Asset;
import org.apache.nifi.asset.AssetManager;
+import org.apache.nifi.controller.flow.FlowManager;
+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.nar.ExtensionManager;
import org.junit.jupiter.api.Test;
@@ -28,12 +32,14 @@ import org.mockito.ArgumentCaptor;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -42,6 +48,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
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.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
@@ -797,12 +804,379 @@ public class TestStandardConnectorRepository {
verify(connector, never()).markInvalid(anyString(), anyString());
}
+ // --- syncConnector tests ---
+
+ @Test
+ public void testSyncConnectorStoppedWithConfigChange() throws Exception {
+ final FlowManager flowManager = mock(FlowManager.class);
+ final StandardConnectorRepository repository =
createRepositoryWithFlowManager(null, flowManager);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(flowManager.createConnector(anyString(), eq("connector-1"),
any(), anyBoolean(), anyBoolean())).thenReturn(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("value1")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED, result.getOutcome());
+ verify(connector).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorStoppedNoConfigChange() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED_CONFIG_UNCHANGED,
result.getOutcome());
+ verify(connector, never()).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorRunningWithConfigChange() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.RUNNING);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("new-value")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED, result.getOutcome());
+ verify(connector).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorDrainingIsRejected() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.DRAINING);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ verify(connector, never()).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorPreparingForUpdateIsRejected() throws
Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+
when(connector.getCurrentState()).thenReturn(ConnectorState.PREPARING_FOR_UPDATE);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ }
+
+ @Test
+ public void testSyncConnectorUpdatingIsRejected() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.UPDATING);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ }
+
+ @Test
+ public void testSyncConnectorUpdateFailedRecovery() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+
when(connector.getCurrentState()).thenReturn(ConnectorState.UPDATE_FAILED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("recovery-value")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED, result.getOutcome());
+ verify(connector).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorUpdatedState() throws Exception {
+ final FlowManager flowManager = mock(FlowManager.class);
+ final StandardConnectorRepository repository =
createRepositoryWithFlowManager(null, flowManager);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(flowManager.createConnector(anyString(), eq("connector-1"),
any(), anyBoolean(), anyBoolean())).thenReturn(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("value1")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED, result.getOutcome());
+ }
+
+ @Test
+ public void testSyncConnectorInheritConfigurationFailureWhenRunning()
throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.RUNNING);
+ doThrow(new FlowUpdateException("Simulated failure"))
+ .when(connector).inheritConfiguration(any(), any(), any());
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("value1")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.FAILED, result.getOutcome());
+ verify(connector).stop(any());
+ }
+
+ @Test
+ public void testSyncConnectorInheritConfigurationFailureWhenStopped()
throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ doThrow(new FlowUpdateException("Simulated failure"))
+ .when(connector).inheritConfiguration(any(), any(), any());
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING,
+ List.of(createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("value1")))));
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.FAILED, result.getOutcome());
+ verify(connector, never()).stop(any());
+ }
+
+ @Test
+ public void testSyncConnectorProviderRejectsSync() throws Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"),
any())).thenReturn(ConnectorSyncDirective.reject());
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ verify(connector, never()).inheritConfiguration(any(), any(), any());
+ }
+
+ @Test
+ public void testSyncConnectorProviderThrowsException() throws Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"), any()))
+ .thenThrow(new
ConnectorConfigurationProviderException("Provider error"));
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.FAILED, result.getOutcome());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ }
+
+ @Test
+ public void testSyncConnectorStartingWaitsForRunning() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithShortTimeout(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState())
+ .thenReturn(ConnectorState.STARTING)
+ .thenReturn(ConnectorState.STARTING)
+ .thenReturn(ConnectorState.RUNNING);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED_CONFIG_UNCHANGED,
result.getOutcome());
+ }
+
+ @Test
+ public void testSyncConnectorStartingTimesOut() throws Exception {
+ final StandardConnectorRepository repository =
createRepositoryWithShortTimeout(null);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STARTING);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ }
+
+ @Test
+ public void testSyncConnectorProviderReturnsRemoveStopsAndRemoves() throws
Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"),
any())).thenReturn(ConnectorSyncDirective.remove());
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final Connector mockExtension = mock(Connector.class);
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState())
+ .thenReturn(ConnectorState.RUNNING)
+ .thenReturn(ConnectorState.RUNNING)
+ .thenReturn(ConnectorState.STOPPED);
+ when(connector.getConnector()).thenReturn(mockExtension);
+
when(connector.stop(any())).thenReturn(CompletableFuture.completedFuture(null));
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REMOVED, result.getOutcome());
+ verify(connector).stop(any());
+ verify(connector).verifyCanDelete();
+ }
+
+ @Test
+ public void testSyncConnectorProviderReturnsRemoveStoppedConnector()
throws Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"),
any())).thenReturn(ConnectorSyncDirective.remove());
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final Connector mockExtension = mock(Connector.class);
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ when(connector.getConnector()).thenReturn(mockExtension);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REMOVED, result.getOutcome());
+ verify(connector, never()).stop(any());
+ verify(connector).verifyCanDelete();
+ }
+
+ @Test
+ public void testSyncConnectorProviderReturnsRemoveForNonExistent() throws
Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"),
any())).thenReturn(ConnectorSyncDirective.remove());
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REMOVED, result.getOutcome());
+ }
+
+ @Test
+ public void
testSyncConnectorProviderReturnsRejectCreatesNodeForNewConnector() throws
Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"),
any())).thenReturn(ConnectorSyncDirective.reject());
+
+ final FlowManager flowManager = mock(FlowManager.class);
+ final StandardConnectorRepository repository =
createRepositoryWithFlowManagerAndProvider(provider, flowManager);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(flowManager.createConnector(anyString(), eq("connector-1"),
any(), anyBoolean(), anyBoolean())).thenReturn(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.REJECTED,
result.getOutcome());
+ verify(flowManager).createConnector(anyString(), eq("connector-1"),
any(), anyBoolean(), anyBoolean());
+ verify(connector).markInvalid(eq("Flow Synchronization Failure"),
anyString());
+ }
+
+ @Test
+ public void testSyncConnectorProviderAllowWithScheduledStateOverride()
throws Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+ when(provider.getSyncDirective(eq("connector-1"), any()))
+ .thenReturn(ConnectorSyncDirective.allow(null,
ScheduledState.ENABLED));
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.RUNNING, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED_CONFIG_UNCHANGED,
result.getOutcome());
+ assertEquals(ScheduledState.ENABLED,
result.getEffectiveScheduledState());
+ }
+
+ @Test
+ public void testSyncConnectorProviderAllowWithWorkingConfig() throws
Exception {
+ final ConnectorConfigurationProvider provider =
mock(ConnectorConfigurationProvider.class);
+
+ final ConnectorWorkingConfiguration providerConfig = new
ConnectorWorkingConfiguration();
+ providerConfig.setName("Provider Name");
+ providerConfig.setWorkingFlowConfiguration(List.of(
+ createVersionedStep("step1", Map.of("prop1",
createStringLiteralRef("provider-value")))));
+ when(provider.getSyncDirective(eq("connector-1"), any()))
+ .thenReturn(ConnectorSyncDirective.allow(providerConfig));
+ final StandardConnectorRepository repository =
createRepositoryWithProvider(provider);
+
+ final ConnectorNode connector =
createConnectorNodeWithEmptyWorkingConfig("connector-1", "Test Connector");
+ when(connector.getCurrentState()).thenReturn(ConnectorState.STOPPED);
+ repository.restoreConnector(connector);
+
+ final VersionedConnector versioned =
createVersionedConnector("connector-1", "Test Connector",
ScheduledState.ENABLED, List.of());
+
+ final ConnectorSyncResult result = repository.syncConnector(versioned);
+
+ assertEquals(ConnectorSyncResult.Outcome.SYNCED, result.getOutcome());
+ verify(connector).setName("Provider Name");
+ verify(connector).inheritConfiguration(any(), any(), any());
+ }
+
// --- Helper Methods ---
private StandardConnectorRepository
createRepositoryWithProviderAndAssetManager(
final ConnectorConfigurationProvider provider, final AssetManager
assetManager) {
final StandardConnectorRepository repository = new
StandardConnectorRepository();
final ConnectorRepositoryInitializationContext initContext =
mock(ConnectorRepositoryInitializationContext.class);
+ when(initContext.getFlowManager()).thenReturn(mock(FlowManager.class));
when(initContext.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
when(initContext.getAssetManager()).thenReturn(assetManager);
when(initContext.getConnectorConfigurationProvider()).thenReturn(provider);
@@ -811,8 +1185,13 @@ public class TestStandardConnectorRepository {
}
private StandardConnectorRepository createRepositoryWithProvider(final
ConnectorConfigurationProvider provider) {
+ return createRepositoryWithFlowManager(provider,
mock(FlowManager.class));
+ }
+
+ private StandardConnectorRepository createRepositoryWithFlowManager(final
ConnectorConfigurationProvider provider, final FlowManager flowManager) {
final StandardConnectorRepository repository = new
StandardConnectorRepository();
final ConnectorRepositoryInitializationContext initContext =
mock(ConnectorRepositoryInitializationContext.class);
+ when(initContext.getFlowManager()).thenReturn(flowManager);
when(initContext.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
when(initContext.getAssetManager()).thenReturn(mock(AssetManager.class));
when(initContext.getConnectorConfigurationProvider()).thenReturn(provider);
@@ -820,6 +1199,11 @@ public class TestStandardConnectorRepository {
return repository;
}
+ private StandardConnectorRepository
createRepositoryWithFlowManagerAndProvider(
+ final ConnectorConfigurationProvider provider, final FlowManager
flowManager) {
+ return createRepositoryWithFlowManager(provider, flowManager);
+ }
+
private ConnectorNode createSimpleConnectorNode(final String id, final
String name) {
final ConnectorNode connector = mock(ConnectorNode.class);
when(connector.getIdentifier()).thenReturn(id);
@@ -874,4 +1258,33 @@ public class TestStandardConnectorRepository {
ref.setValue(value);
return ref;
}
+
+ private VersionedConnector createVersionedConnector(final String id, final
String name, final ScheduledState scheduledState,
+ final
List<VersionedConfigurationStep> activeConfig) {
+ final VersionedConnector vc = new VersionedConnector();
+ vc.setInstanceIdentifier(id);
+ vc.setName(name);
+ vc.setScheduledState(scheduledState);
+ vc.setActiveFlowConfiguration(activeConfig);
+ vc.setWorkingFlowConfiguration(activeConfig);
+ final Bundle bundle = new Bundle();
+ bundle.setGroup("org.apache.nifi");
+ bundle.setArtifact("nifi-test-connector-nar");
+ bundle.setVersion("2.0.0");
+ vc.setBundle(bundle);
+ vc.setType("org.apache.nifi.test.TestConnector");
+ return vc;
+ }
+
+ private StandardConnectorRepository createRepositoryWithShortTimeout(final
ConnectorConfigurationProvider provider) {
+ final StandardConnectorRepository repository = new
StandardConnectorRepository();
+ final ConnectorRepositoryInitializationContext initContext =
mock(ConnectorRepositoryInitializationContext.class);
+ when(initContext.getFlowManager()).thenReturn(mock(FlowManager.class));
+
when(initContext.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
+
when(initContext.getAssetManager()).thenReturn(mock(AssetManager.class));
+
when(initContext.getConnectorConfigurationProvider()).thenReturn(provider);
+
when(initContext.getConnectorSyncTimeout()).thenReturn(Duration.ofSeconds(5));
+ repository.initialize(initContext);
+ return repository;
+ }
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
index 71173f8a89f..47b5c2c02c1 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
@@ -20,6 +20,7 @@ import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.connector.ConnectorNode;
import org.apache.nifi.components.connector.ConnectorRepository;
+import org.apache.nifi.components.connector.ConnectorSyncResult;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ReportingTaskNode;
import org.apache.nifi.controller.SnippetManager;
@@ -293,7 +294,7 @@ class VersionedFlowSynchronizerTest {
}
@Test
- void testSyncInheritConnectorStartRoutedThroughFlowController() {
+ void testSyncInheritConnectorDelegatesSyncToConnectorRepository() {
setRootGroup();
final String connectorId = UUID.randomUUID().toString();
@@ -307,16 +308,17 @@ class VersionedFlowSynchronizerTest {
versionedConnector.setScheduledState(ScheduledState.RUNNING);
final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
- final ConnectorNode connectorNode = mock(ConnectorNode.class);
-
when(connectorRepository.getConnector(eq(connectorId))).thenReturn(null).thenReturn(connectorNode);
- when(flowManager.createConnector(eq(connectorType), eq(connectorId),
any(), eq(false), eq(true))).thenReturn(connectorNode);
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(versionedConnector))
+ .thenReturn(ConnectorSyncResult.synced(syncedNode,
ScheduledState.RUNNING));
setFlowController(connectorRepository);
when(versionedDataflow.getConnectors()).thenReturn(List.of(versionedConnector));
versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
- verify(flowController).startConnector(connectorNode);
+ verify(connectorRepository).syncConnector(versionedConnector);
+ verify(flowController).startConnector(syncedNode);
}
@Test
@@ -334,16 +336,123 @@ class VersionedFlowSynchronizerTest {
versionedConnector.setScheduledState(ScheduledState.ENABLED);
final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
- final ConnectorNode connectorNode = mock(ConnectorNode.class);
-
when(connectorRepository.getConnector(eq(connectorId))).thenReturn(null).thenReturn(connectorNode);
- when(flowManager.createConnector(eq(connectorType), eq(connectorId),
any(), eq(false), eq(true))).thenReturn(connectorNode);
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(versionedConnector))
+
.thenReturn(ConnectorSyncResult.syncedConfigUnchanged(syncedNode,
ScheduledState.ENABLED));
setFlowController(connectorRepository);
when(versionedDataflow.getConnectors()).thenReturn(List.of(versionedConnector));
versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
+ verify(connectorRepository).syncConnector(versionedConnector);
verify(flowController, never()).startConnector(any());
- verify(connectorRepository).stopConnector(connectorNode);
+ verify(connectorRepository).stopConnector(syncedNode);
+ }
+
+ @Test
+ void testSyncOrphanConnectorIsRemoved() {
+ setRootGroup();
+
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+
+ // The proposed flow has one connector, but the local repo has a
different one (orphan)
+ final VersionedConnector proposedConnector = new VersionedConnector();
+ proposedConnector.setInstanceIdentifier("proposed-connector-id");
+ proposedConnector.setName("Proposed Connector");
+ proposedConnector.setType("org.apache.nifi.connectors.TestConnector");
+ proposedConnector.setBundle(CORE_BUNDLE);
+ proposedConnector.setScheduledState(ScheduledState.ENABLED);
+
+ final ConnectorNode orphanConnector = mock(ConnectorNode.class);
+
org.mockito.Mockito.lenient().when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
+ when(connectorRepository.stopConnector(orphanConnector))
+
.thenReturn(java.util.concurrent.CompletableFuture.completedFuture(null));
+
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(proposedConnector))
+
.thenReturn(ConnectorSyncResult.syncedConfigUnchanged(syncedNode,
ScheduledState.ENABLED));
+ when(connectorRepository.stopConnector(syncedNode))
+
.thenReturn(java.util.concurrent.CompletableFuture.completedFuture(null));
+
+ setFlowController(connectorRepository);
+
when(connectorRepository.getConnectors()).thenReturn(List.of(orphanConnector));
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(proposedConnector));
+
+ versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
+
+ verify(connectorRepository).syncConnector(proposedConnector);
+ verify(connectorRepository).stopConnector(orphanConnector);
+ verify(connectorRepository).removeConnector("orphan-connector-id");
+ }
+
+ @Test
+ void testSyncOrphanConnectorNotRemovedWhenInProposedFlow() {
+ setRootGroup();
+
+ final String connectorId = UUID.randomUUID().toString();
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+
+ final ConnectorNode existingConnector = mock(ConnectorNode.class);
+
org.mockito.Mockito.lenient().when(existingConnector.getIdentifier()).thenReturn(connectorId);
+
+ final VersionedConnector versionedConnector = new VersionedConnector();
+ versionedConnector.setInstanceIdentifier(connectorId);
+ versionedConnector.setName("Test Connector");
+ versionedConnector.setType("org.apache.nifi.connectors.TestConnector");
+ versionedConnector.setBundle(CORE_BUNDLE);
+ versionedConnector.setScheduledState(ScheduledState.ENABLED);
+
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(versionedConnector))
+
.thenReturn(ConnectorSyncResult.syncedConfigUnchanged(syncedNode,
ScheduledState.ENABLED));
+ when(connectorRepository.stopConnector(syncedNode))
+
.thenReturn(java.util.concurrent.CompletableFuture.completedFuture(null));
+
+ setFlowController(connectorRepository);
+
when(connectorRepository.getConnectors()).thenReturn(List.of(existingConnector));
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(versionedConnector));
+
+ versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
+
+ verify(connectorRepository, never()).removeConnector(any());
+ verify(connectorRepository).syncConnector(versionedConnector);
+ }
+
+ @Test
+ void testSyncOrphanRemovalFailureMarksInvalid() {
+ setRootGroup();
+
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+
+ // The proposed flow has one connector; the orphan is a different one
+ final VersionedConnector proposedConnector = new VersionedConnector();
+ proposedConnector.setInstanceIdentifier("proposed-connector-id");
+ proposedConnector.setName("Proposed Connector");
+ proposedConnector.setType("org.apache.nifi.connectors.TestConnector");
+ proposedConnector.setBundle(CORE_BUNDLE);
+ proposedConnector.setScheduledState(ScheduledState.ENABLED);
+
+ final ConnectorNode orphanConnector = mock(ConnectorNode.class);
+
org.mockito.Mockito.lenient().when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
+
+ final java.util.concurrent.CompletableFuture<Void> failedFuture = new
java.util.concurrent.CompletableFuture<>();
+ failedFuture.completeExceptionally(new RuntimeException("Stop
failed"));
+
when(connectorRepository.stopConnector(orphanConnector)).thenReturn(failedFuture);
+
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(proposedConnector))
+
.thenReturn(ConnectorSyncResult.syncedConfigUnchanged(syncedNode,
ScheduledState.ENABLED));
+ when(connectorRepository.stopConnector(syncedNode))
+
.thenReturn(java.util.concurrent.CompletableFuture.completedFuture(null));
+
+ setFlowController(connectorRepository);
+
when(connectorRepository.getConnectors()).thenReturn(List.of(orphanConnector));
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(proposedConnector));
+
+ versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
+
+ verify(connectorRepository).syncConnector(proposedConnector);
+ verify(orphanConnector).markInvalid(eq("Flow Synchronization
Failure"), any());
}
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
index b07646dd179..cc22e9fa271 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorIT.java
@@ -201,6 +201,88 @@ public class ClusteredConnectorIT extends ConnectorCrudIT {
}
+ @Test
+ public void testRunningConnectorSyncedOnReconnect() throws
NiFiClientException, IOException, InterruptedException {
+ // Create and start a connector
+ final ConnectorEntity connector =
getClientUtil().createConnector("NopConnector");
+ assertNotNull(connector);
+ getClientUtil().configureConnector(connector, "Ignored Step",
Map.of("Ignored Property", "Test Value"));
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().startConnector(connector);
+ logger.info("Created and started connector [{}]", connector.getId());
+
+ // Verify running on both nodes
+ switchClientToNode(1);
+ ConnectorEntity node1Connector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ assertEquals("RUNNING", node1Connector.getComponent().getState());
+
+ switchClientToNode(2);
+ ConnectorEntity node2Connector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ assertEquals("RUNNING", node2Connector.getComponent().getState());
+ logger.info("Verified connector is RUNNING on both nodes");
+
+ // Disconnect node 2
+ switchClientToNode(1);
+ disconnectNode(2);
+ logger.info("Disconnected Node 2");
+
+ // Reconnect node 2
+ reconnectNode(2);
+ waitForAllNodesConnected();
+ logger.info("Reconnected Node 2");
+
+ // Verify the connector is running on node 2 after rejoin
+ switchClientToNode(2);
+ waitFor(() -> {
+ try {
+ final ConnectorEntity rejoinedConnector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ return
"RUNNING".equals(rejoinedConnector.getComponent().getState());
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+
+ final ConnectorEntity finalNode2Connector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ assertEquals("RUNNING", finalNode2Connector.getComponent().getState());
+ assertActiveConfigurationValue(finalNode2Connector, "Ignored
Property", "Test Value");
+ logger.info("Verified connector is RUNNING on Node 2 after rejoin with
correct configuration");
+ }
+
+ @Test
+ public void testNewConnectorSyncedOnReconnect() throws
NiFiClientException, IOException, InterruptedException {
+ // Disconnect node 2
+ disconnectNode(2);
+ logger.info("Disconnected Node 2");
+
+ // Create a connector on node 1 while node 2 is disconnected
+ final ConnectorEntity connector =
getClientUtil().createConnector("NopConnector");
+ assertNotNull(connector);
+ getClientUtil().configureConnector(connector, "Ignored Step",
Map.of("Ignored Property", "Created While Disconnected"));
+ getClientUtil().applyConnectorUpdate(connector);
+ logger.info("Created connector [{}] on Node 1 while Node 2 is
disconnected", connector.getId());
+
+ // Reconnect node 2
+ reconnectNode(2);
+ waitForAllNodesConnected();
+ logger.info("Reconnected Node 2");
+
+ // Verify the connector exists on node 2 after rejoin
+ switchClientToNode(2);
+ waitFor(() -> {
+ try {
+
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ return true;
+ } catch (final Exception e) {
+ return false;
+ }
+ });
+
+ final ConnectorEntity node2Connector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connector.getId());
+ assertNotNull(node2Connector);
+ assertActiveConfigurationValue(node2Connector, "Ignored Property",
"Created While Disconnected");
+ logger.info("Verified new connector exists on Node 2 after rejoin with
correct configuration");
+ }
+
private void assertActiveConfigurationValue(final ConnectorEntity
connector, final String propertyName, final String expectedValue) {
final String actualValue =
getConfigurationValue(connector.getComponent().getActiveConfiguration(),
propertyName);
assertEquals(expectedValue, actualValue, "Active configuration
property '" + propertyName + "' did not match expected value");
