Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
markap14 merged PR #11079: URL: https://github.com/apache/nifi/pull/11079 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025786261
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java:
##
@@ -91,6 +107,288 @@ public void restoreConnector(final ConnectorNode
connector) {
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.verifySyncable(connectorId,
proposedScheduledState);
+} catch (final Exception e) {
+logger.error("Configuration provider threw exception during
verifySyncable 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;
+}
+} 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.remove(connectorId);
Review Comment:
Ok added purge
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025758999
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java:
##
@@ -1042,36 +1042,54 @@ private Set getAssetIds(final Parameter
parameter) {
}
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 proposedConnectorIds = new HashSet<>();
if (dataflow.getConnectors() != null) {
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());
+// Ensure the connector exists locally before any provider
interaction so that
+// if verifySyncable or any subsequent step fails, we have a
local node to mark invalid.
+ConnectorNode connectorNode =
connectorRepository.getConnector(versionedConnector.getInstanceIdentifier());
+final boolean isNewConnector = connectorNode == null;
+if (isNewConnector) {
+final BundleCoordinate coordinate =
createBundleCoordinate(extensionManager, versionedConnector.getBundle(),
versionedConnector.getType());
+connectorNode =
flowController.getFlowManager().createConnector(
+versionedConnector.getType(),
versionedConnector.getInstanceIdentifier(), coordinate, false, true);
+connectorRepository.restoreConnector(connectorNode);
+}
-addConnector(versionedConnector, connectorRepository,
flowController.getFlowManager());
-} else if (isConnectorConfigurationUpdated(existingConnector,
versionedConnector)) {
-logger.info("{} configuration has changed, updating
configuration", existingConnector);
-updateConnector(versionedConnector, connectorRepository);
-} else {
-logger.debug("{} configuration is up to date, no update
necessary", existingConnector);
+try {
+
connectorRepository.verifySyncable(versionedConnector.getInstanceIdentifier());
Review Comment:
OBE
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025758383
##
nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorConfigurationProvider.java:
##
@@ -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.
+ *
+ * When the directive action is {@link
ConnectorSyncDirective.Action#ALLOW}, the
+ * directive may optionally include:
+ *
+ * A {@link ConnectorWorkingConfiguration} with the provider's
working config and name
+ * (overriding the potentially stale values from the versioned
flow)
+ * A {@link ScheduledState} override (correcting stale run intent
from the versioned flow)
+ *
+ *
+ * This method combines the verify and load operations into a single
call to avoid
+ * redundant round-trips to the external store.
+ *
+ * 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.
+ *
+ * @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 verifySyncable(final String connectorId,
final ScheduledState proposedScheduledState) {
Review Comment:
I like getSyncDirective
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025756865
##
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java:
##
@@ -47,4 +49,14 @@ default ConnectorConfigurationProvider
getConnectorConfigurationProvider() {
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 15 minutes
+ */
+default Duration getConnectorSyncTimeout() {
+return Duration.ofMinutes(15);
Review Comment:
ha there are some waitForState operations that could take a while, but I
agree. How do you feel about 5 minutes?
##
nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncDirective.java:
##
@@ -0,0 +1,149 @@
+/*
+ * 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#verifySyncable(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.A
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025752702
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java:
##
@@ -91,6 +107,288 @@ public void restoreConnector(final ConnectorNode
connector) {
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.verifySyncable(connectorId,
proposedScheduledState);
+} catch (final Exception e) {
+logger.error("Configuration provider threw exception during
verifySyncable 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;
+}
+} 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.remove(connectorId);
Review Comment:
Yeah I think I addressed most of this in my last commit, but let me check
that we are also purging data.
##
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncResult.java:
##
@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+/**
+ * Result of a connector synchronization attempt during flow inheritance.
+ */
+public enum ConnectorSyncResult {
+
+/**
+ * Configuration was applied and the connector's run state was updated
+ * to match the proposed ScheduledState.
+ */
+SYNCED,
+
+/**
+ * Configuration was already up to date; the connector's run state was
+ * updated if it differed from the proposed ScheduledState.
+ */
+SYNCED_NO_CHANGES,
Review Comment:
Yep
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3025751179
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java:
##
@@ -39,34 +45,44 @@
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 DEFAULT_SYNC_TIMEOUT =
Duration.ofMinutes(15);
+private static final Duration SYNC_POLL_INTERVAL = Duration.ofSeconds(2);
private final Map 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 = DEFAULT_SYNC_TIMEOUT;
Review Comment:
Yep
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
markap14 commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3023935120
##
nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorRepositoryInitializationContext.java:
##
@@ -47,4 +49,14 @@ default ConnectorConfigurationProvider
getConnectorConfigurationProvider() {
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 15 minutes
+ */
+default Duration getConnectorSyncTimeout() {
+return Duration.ofMinutes(15);
Review Comment:
This feels like a rather excessive default, no?
##
nifi-framework-api/src/main/java/org/apache/nifi/components/connector/ConnectorSyncDirective.java:
##
@@ -0,0 +1,149 @@
+/*
+ * 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#verifySyncable(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);
+}
+
+/**
+ * Retu
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
kevdoran commented on PR #11079: URL: https://github.com/apache/nifi/pull/11079#issuecomment-4172050540 Thanks for the review, @pvillard31. Prompted by your comments and some other discussions, I took a step back and came up with a fundamentally different approach. I have a backup of the old approach on a branch if we want to revisit that, but this PR has been updated with my current solution. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] NIFI-15751 Improve Nifi Cluster Sync with Connectors [nifi]
pvillard31 commented on code in PR #11079:
URL: https://github.com/apache/nifi/pull/11079#discussion_r3012234375
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java:
##
@@ -101,6 +129,7 @@ public void removeConnector(final String connectorId) {
connectorNode.verifyCanDelete();
if (configurationProvider != null) {
+configurationProvider.verifyDelete(connectorId);
Review Comment:
`verifyDelete` is already called by `StandardConnectorDAO.verifyDelete()`
before this method is invoked, is it necessary to have it here as well?
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorRepository.java:
##
@@ -451,29 +480,35 @@ private void syncFromProvider(final ConnectorNode
connector) {
}
final String connectorId = connector.getIdentifier();
+
+final Optional metadata =
configurationProvider.loadMetadata(connectorId);
+metadata.ifPresent(m -> {
+if (m.getName() != null) {
+connector.setName(m.getName());
+}
+});
Review Comment:
Metadata and configuration are now loaded in two separate calls. The
previous single `load()` was atomic. If the external store changes between
these two calls, the connector could get metadata from one version and
configuration from another. Is that acceptable?
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java:
##
@@ -1042,36 +1042,54 @@ private Set getAssetIds(final Parameter
parameter) {
}
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 proposedConnectorIds = new HashSet<>();
if (dataflow.getConnectors() != null) {
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());
+// Ensure the connector exists locally before any provider
interaction so that
+// if verifySyncable or any subsequent step fails, we have a
local node to mark invalid.
+ConnectorNode connectorNode =
connectorRepository.getConnector(versionedConnector.getInstanceIdentifier());
+final boolean isNewConnector = connectorNode == null;
+if (isNewConnector) {
+final BundleCoordinate coordinate =
createBundleCoordinate(extensionManager, versionedConnector.getBundle(),
versionedConnector.getType());
+connectorNode =
flowController.getFlowManager().createConnector(
+versionedConnector.getType(),
versionedConnector.getInstanceIdentifier(), coordinate, false, true);
+connectorRepository.restoreConnector(connectorNode);
+}
-addConnector(versionedConnector, connectorRepository,
flowController.getFlowManager());
-} else if (isConnectorConfigurationUpdated(existingConnector,
versionedConnector)) {
-logger.info("{} configuration has changed, updating
configuration", existingConnector);
-updateConnector(versionedConnector, connectorRepository);
-} else {
-logger.debug("{} configuration is up to date, no update
necessary", existingConnector);
+try {
+
connectorRepository.verifySyncable(versionedConnector.getInstanceIdentifier());
Review Comment:
Shouldn't this be even before the connectorRepository.getConnector() a few
lines above?
##
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java:
##
@@ -1162,18 +1180,15 @@ private boolean equals(final
VersionedConnectorValueReference version
