This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 73eb46b01a8 NIFI-16308 Prevent concurrent Connector starts (#11638)
73eb46b01a8 is described below
commit 73eb46b01a853cd9b8c7795f0c3370c798b422be
Author: Mark Payne <[email protected]>
AuthorDate: Mon Sep 7 14:29:16 2026 -0400
NIFI-16308 Prevent concurrent Connector starts (#11638)
---
.../connector/ConnectorStateTransition.java | 2 +-
.../connector/StandardConnectorNode.java | 55 +++++----
.../StandardConnectorStateTransition.java | 68 +++++++----
.../connector/TestStandardConnectorNode.java | 125 ++++++++++++++++++++-
.../TestStandardConnectorStateTransition.java | 38 +++++++
5 files changed, 240 insertions(+), 48 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorStateTransition.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorStateTransition.java
index a9693f8eafc..c0c965946b0 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorStateTransition.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/components/connector/ConnectorStateTransition.java
@@ -64,7 +64,7 @@ public interface ConnectorStateTransition {
void setCurrentState(ConnectorState newState);
/**
- * Registers a future to be completed when the connector transitions to
the RUNNING state.
+ * Completes a future immediately when the connector is running, or
registers it to be completed when the connector transitions to the RUNNING
state.
* This method is thread-safe and handles internal synchronization.
*
* @param future the CompletableFuture to complete when the connector
starts
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
index 1e0ae929d4c..06b87484e8e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorNode.java
@@ -679,30 +679,41 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
verifyCanStart();
- final ConnectorState currentState = getCurrentState();
- switch (currentState) {
- case STARTING -> {
- logger.debug("{} is already starting; adding future to
pending start futures", this);
- stateTransition.addPendingStartFuture(startCompleteFuture);
- }
- case RUNNING -> {
- logger.debug("{} is already {}; will not attempt to
start", this, currentState);
- startCompleteFuture.complete(null);
- }
- case STOPPING -> {
- // We have set the Desired State to RUNNING so when the
Connector fully stops, it will be started again automatically
- logger.info("{} is currently stopping so will not trigger
Connector to start until it has fully stopped", this);
- stateTransition.addPendingStartFuture(startCompleteFuture);
- }
- case STOPPED, PREPARING_FOR_UPDATE, UPDATED -> {
- stateTransition.setCurrentState(ConnectorState.STARTING);
- scheduler.schedule(() -> startComponent(scheduler,
startCompleteFuture), 0, TimeUnit.SECONDS);
- }
- default -> {
- logger.warn("{} is in state {} and cannot be started",
this, currentState);
- stateTransition.addPendingStartFuture(startCompleteFuture);
+ boolean startScheduled = false;
+ while (!startScheduled) {
+ final ConnectorState currentState = getCurrentState();
+ switch (currentState) {
+ case STARTING -> {
+ logger.debug("{} is already starting; adding future to
pending start futures", this);
+
stateTransition.addPendingStartFuture(startCompleteFuture);
+ return;
+ }
+ case RUNNING -> {
+ logger.debug("{} is already {}; will not attempt to
start", this, currentState);
+ startCompleteFuture.complete(null);
+ return;
+ }
+ case STOPPING -> {
+ // We have set the Desired State to RUNNING so when
the Connector fully stops, it will be started again automatically
+ logger.info("{} is currently stopping so will not
trigger Connector to start until it has fully stopped", this);
+
stateTransition.addPendingStartFuture(startCompleteFuture);
+ return;
+ }
+ case STOPPED, PREPARING_FOR_UPDATE, UPDATED -> {
+ startScheduled =
stateTransition.trySetCurrentState(currentState, ConnectorState.STARTING);
+ if (startScheduled) {
+ logger.info("Starting {}", this);
+ }
+ }
+ default -> {
+ logger.warn("{} is in state {} and cannot be started",
this, currentState);
+
stateTransition.addPendingStartFuture(startCompleteFuture);
+ return;
+ }
}
}
+
+ scheduler.schedule(() -> startComponent(scheduler,
startCompleteFuture), 0, TimeUnit.SECONDS);
} catch (final Exception e) {
logger.error("Failed to start {}", this, e);
startCompleteFuture.completeExceptionally(e);
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorStateTransition.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorStateTransition.java
index 9cfd4cccb38..3e45c73bc3d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorStateTransition.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/StandardConnectorStateTransition.java
@@ -56,25 +56,47 @@ public class StandardConnectorStateTransition implements
ConnectorStateTransitio
@Override
public boolean trySetCurrentState(final ConnectorState expectedState,
final ConnectorState newState) {
- final boolean changed = currentState.compareAndSet(expectedState,
newState);
- if (changed) {
- logger.info("Transitioned current state for {} from {} to {}",
componentDescription, expectedState, newState);
- completeFuturesForStateTransition(newState);
+ final List<CompletableFuture<Void>> futuresToComplete;
+ synchronized (this) {
+ final boolean changed = currentState.compareAndSet(expectedState,
newState);
+ if (!changed) {
+ return false;
+ }
+
+ futuresToComplete = removePendingFutures(newState);
}
- return changed;
+ logger.info("Transitioned current state for {} from {} to {}",
componentDescription, expectedState, newState);
+ completeFutures(futuresToComplete, newState);
+ return true;
}
@Override
public void setCurrentState(final ConnectorState newState) {
- final ConnectorState oldState = currentState.getAndSet(newState);
+ final ConnectorState oldState;
+ final List<CompletableFuture<Void>> futuresToComplete;
+ synchronized (this) {
+ oldState = currentState.getAndSet(newState);
+ futuresToComplete = removePendingFutures(newState);
+ }
+
logger.info("Transitioned current state for {} from {} to {}",
componentDescription, oldState, newState);
- completeFuturesForStateTransition(newState);
+ completeFutures(futuresToComplete, newState);
}
@Override
- public synchronized void addPendingStartFuture(final
CompletableFuture<Void> future) {
- pendingStartFutures.add(future);
+ public void addPendingStartFuture(final CompletableFuture<Void> future) {
+ final boolean completeImmediately;
+ synchronized (this) {
+ completeImmediately = currentState.get() == ConnectorState.RUNNING;
+ if (!completeImmediately) {
+ pendingStartFutures.add(future);
+ }
+ }
+
+ if (completeImmediately) {
+ future.complete(null);
+ }
}
@Override
@@ -82,31 +104,29 @@ public class StandardConnectorStateTransition implements
ConnectorStateTransitio
pendingStopFutures.add(future);
}
- private synchronized void completeFuturesForStateTransition(final
ConnectorState newState) {
+ private List<CompletableFuture<Void>> removePendingFutures(final
ConnectorState newState) {
if (newState == ConnectorState.RUNNING) {
final List<CompletableFuture<Void>> futuresToComplete = new
ArrayList<>(pendingStartFutures);
pendingStartFutures.clear();
-
- for (final CompletableFuture<Void> future : futuresToComplete) {
- future.complete(null);
- }
-
- if (!futuresToComplete.isEmpty()) {
- logger.debug("Completed {} pending start futures for {}",
futuresToComplete.size(), componentDescription);
- }
+ return futuresToComplete;
}
if (newState == ConnectorState.STOPPED) {
final List<CompletableFuture<Void>> futuresToComplete = new
ArrayList<>(pendingStopFutures);
pendingStopFutures.clear();
+ return futuresToComplete;
+ }
- for (final CompletableFuture<Void> future : futuresToComplete) {
- future.complete(null);
- }
+ return List.of();
+ }
- if (!futuresToComplete.isEmpty()) {
- logger.debug("Completed {} pending stop futures for {}",
futuresToComplete.size(), componentDescription);
- }
+ private void completeFutures(final List<CompletableFuture<Void>> futures,
final ConnectorState newState) {
+ for (final CompletableFuture<Void> future : futures) {
+ future.complete(null);
+ }
+
+ if (!futures.isEmpty()) {
+ logger.debug("Completed {} pending futures for {} on transition to
{}", futures.size(), componentDescription, newState);
}
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
index 767c20a5abc..f96b2e9af75 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorNode.java
@@ -65,9 +65,12 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -231,6 +234,32 @@ public class TestStandardConnectorNode {
assertTrue(startFuture2.isDone());
}
+ @Test
+ @Timeout(value = 10, unit = TimeUnit.SECONDS)
+ public void testConcurrentStartRequestsInvokeConnectorStartOnce() throws
Exception {
+ final ConcurrentStartConnector connector = new
ConcurrentStartConnector();
+ final CoordinatedConnectorStateTransition stateTransition = new
CoordinatedConnectorStateTransition();
+ final StandardConnectorNode connectorNode =
createConnectorNode(connector, stateTransition);
+ connector.blockValidation();
+
+ try (final ExecutorService executor = Executors.newFixedThreadPool(2))
{
+ final CompletableFuture<Future<Void>> firstRequest =
CompletableFuture.supplyAsync(() -> connectorNode.start(scheduler), executor);
+ final CompletableFuture<Future<Void>> secondRequest =
CompletableFuture.supplyAsync(() -> connectorNode.start(scheduler), executor);
+
+ assertTrue(connector.awaitValidationRequests(5, TimeUnit.SECONDS));
+ stateTransition.coordinateNextStateReads(2);
+ connector.releaseValidation();
+
+ final Future<Void> firstStart = firstRequest.get(5,
TimeUnit.SECONDS);
+ final Future<Void> secondStart = secondRequest.get(5,
TimeUnit.SECONDS);
+ firstStart.get(5, TimeUnit.SECONDS);
+ secondStart.get(5, TimeUnit.SECONDS);
+ }
+
+ assertEquals(1, connector.getStartInvocations());
+ assertEquals(ConnectorState.RUNNING, connectorNode.getCurrentState());
+ }
+
@Test
public void testVerifyCanDeleteWhenStopped() throws FlowUpdateException {
final StandardConnectorNode connectorNode = createConnectorNode();
@@ -1365,8 +1394,20 @@ public class TestStandardConnectorNode {
return createConnectorNode(connector, defaultSecretsManager);
}
+ private StandardConnectorNode createConnectorNode(final Connector
connector, final ConnectorStateTransition stateTransition) throws
FlowUpdateException {
+ final SecretsManager defaultSecretsManager =
mock(SecretsManager.class);
+ when(defaultSecretsManager.getAllSecrets()).thenReturn(List.of());
+
when(defaultSecretsManager.getSecrets(anySet())).thenReturn(Collections.emptyMap());
+ when(defaultSecretsManager.getSecrets(anySet(),
anyBoolean())).thenReturn(Collections.emptyMap());
+ return createConnectorNode(connector, defaultSecretsManager,
stateTransition);
+ }
+
private StandardConnectorNode createConnectorNode(final Connector
connector, final SecretsManager initializedSecretsManager) throws
FlowUpdateException {
- final ConnectorStateTransition stateTransition = new
StandardConnectorStateTransition("TestConnectorNode");
+ return createConnectorNode(connector, initializedSecretsManager, new
StandardConnectorStateTransition("TestConnectorNode"));
+ }
+
+ private StandardConnectorNode createConnectorNode(final Connector
connector, final SecretsManager initializedSecretsManager,
+ final ConnectorStateTransition stateTransition) throws
FlowUpdateException {
final ConnectorValidationTrigger validationTrigger = new
SynchronousConnectorValidationTrigger();
final StandardConnectorNode node = new StandardConnectorNode(
"test-connector-id",
@@ -1392,6 +1433,88 @@ public class TestStandardConnectorNode {
return node;
}
+ private static class ConcurrentStartConnector extends SleepingConnector {
+ private final CountDownLatch validationRequests = new
CountDownLatch(2);
+ private final CountDownLatch validationRelease = new CountDownLatch(1);
+ private final AtomicInteger startInvocations = new AtomicInteger();
+ private volatile boolean validationBlocked;
+
+ private ConcurrentStartConnector() {
+ super(Duration.ZERO);
+ }
+
+ @Override
+ public List<ValidationResult> validate(final FlowContext flowContext,
final ConnectorValidationContext connectorValidationContext) {
+ if (validationBlocked) {
+ validationRequests.countDown();
+ try {
+ validationRelease.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while waiting
to release validation", e);
+ }
+ }
+
+ return List.of();
+ }
+
+ @Override
+ public void start(final FlowContext activeContext) {
+ startInvocations.incrementAndGet();
+ }
+
+ private void blockValidation() {
+ validationBlocked = true;
+ }
+
+ private boolean awaitValidationRequests(final long timeout, final
TimeUnit timeUnit) throws InterruptedException {
+ return validationRequests.await(timeout, timeUnit);
+ }
+
+ private void releaseValidation() {
+ validationRelease.countDown();
+ }
+
+ private int getStartInvocations() {
+ return startInvocations.get();
+ }
+ }
+
+ private static class CoordinatedConnectorStateTransition extends
StandardConnectorStateTransition {
+ private volatile CountDownLatch coordinatedStateReads;
+
+ private CoordinatedConnectorStateTransition() {
+ super("TestConnectorNode");
+ }
+
+ @Override
+ public ConnectorState getCurrentState() {
+ final CountDownLatch stateReads = coordinatedStateReads;
+ final ConnectorState currentState = super.getCurrentState();
+ if (stateReads != null) {
+ stateReads.countDown();
+ try {
+ if (!stateReads.await(5, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting for
coordinated state reads");
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while waiting
for coordinated state reads", e);
+ } finally {
+ if (stateReads.getCount() == 0) {
+ coordinatedStateReads = null;
+ }
+ }
+ }
+
+ return currentState;
+ }
+
+ private void coordinateNextStateReads(final int count) {
+ coordinatedStateReads = new CountDownLatch(count);
+ }
+ }
+
private static class SynchronousConnectorValidationTrigger implements
ConnectorValidationTrigger {
@Override
public void triggerAsync(final ConnectorNode connector) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorStateTransition.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorStateTransition.java
new file mode 100644
index 00000000000..aaf29f01676
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/TestStandardConnectorStateTransition.java
@@ -0,0 +1,38 @@
+/*
+ * 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.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestStandardConnectorStateTransition {
+
+ @Test
+ void testAddPendingStartFutureWhenRunningCompletesFuture() {
+ final StandardConnectorStateTransition stateTransition = new
StandardConnectorStateTransition("Test Connector");
+ stateTransition.setCurrentState(ConnectorState.RUNNING);
+ final CompletableFuture<Void> startFuture = new CompletableFuture<>();
+
+ stateTransition.addPendingStartFuture(startFuture);
+
+ assertTrue(startFuture.isDone());
+ }
+}
