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 bc53d07bb93 NIFI-16262 Fixed Connector stop handling for multiple
callers (#11597)
bc53d07bb93 is described below
commit bc53d07bb93dda7a1b168107c0525a132364f151
Author: David Handermann <[email protected]>
AuthorDate: Mon Aug 31 04:04:51 2026 -0500
NIFI-16262 Fixed Connector stop handling for multiple callers (#11597)
---
.../connector/StandardConnectorNode.java | 41 +++++-
.../connector/TestStandardConnectorNode.java | 161 ++++++++++++++++++++-
2 files changed, 195 insertions(+), 7 deletions(-)
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 552a95f90ea..1e0ae929d4c 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
@@ -88,7 +88,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -737,6 +736,13 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
}
stateUpdated = stateTransition.trySetCurrentState(currentState,
ConnectorState.STOPPING);
+
+ // Check current state for existing start request in progress
+ if (stateUpdated && currentState == ConnectorState.STARTING) {
+ logger.info("{} is currently starting so will not stop the
Connector until the start has completed", this);
+ stateTransition.addPendingStopFuture(stopCompleteFuture);
+ return stopCompleteFuture;
+ }
}
scheduler.schedule(() -> stopComponent(scheduler, stopCompleteFuture),
0, TimeUnit.SECONDS);
@@ -882,25 +888,48 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
}
}
- private void startComponent(final ScheduledExecutorService scheduler,
final CompletableFuture<Void> startCompleteFuture) {
+ private void startComponent(final FlowEngine scheduler, final
CompletableFuture<Void> startCompleteFuture) {
logger.debug("Starting component for {}", this);
final ConnectorState desiredState = getDesiredState();
if (desiredState != ConnectorState.RUNNING) {
logger.info("Will not start {} because the desired state is no
longer RUNNING but is now {}", this, desiredState);
+ completeDeferredStop(scheduler);
return;
}
try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(extensionManager,
connectorDetails.getConnector().getClass(), getIdentifier())) {
connectorDetails.getConnector().start(activeFlowContext);
} catch (final Exception e) {
- logger.error("Failed to start {}. Will try again in 10 seconds",
this, e);
- scheduler.schedule(() -> startComponent(scheduler,
startCompleteFuture), 10, TimeUnit.SECONDS);
+ if (getCurrentState() == ConnectorState.STOPPING) {
+ logger.error("Failed to start {} and a stop has since been
requested, so the Connector will be stopped instead of started", this, e);
+ completeDeferredStop(scheduler);
+ } else {
+ logger.error("Failed to start {} retrying in 10 seconds",
this, e);
+ scheduler.schedule(() -> startComponent(scheduler,
startCompleteFuture), 10, TimeUnit.SECONDS);
+ }
+
return;
}
- stateTransition.setCurrentState(ConnectorState.RUNNING);
+ // A stop requested while the Connector was starting transitioned the
current state away from STARTING, so the
+ // Connector may be reported as RUNNING only if it is still STARTING.
Otherwise, this thread owns the stop that
+ // was deferred while the start was in flight.
+ final boolean transitionedToRunning =
stateTransition.trySetCurrentState(ConnectorState.STARTING,
ConnectorState.RUNNING);
startCompleteFuture.complete(null);
- logger.info("Successfully started {}", this);
+
+ if (transitionedToRunning) {
+ logger.info("Successfully started {}", this);
+ } else {
+ logger.info("Started {} but its current state is now {} so it will
not be reported as RUNNING", this, getCurrentState());
+ completeDeferredStop(scheduler);
+ }
+ }
+
+ private void completeDeferredStop(final FlowEngine scheduler) {
+ if (getCurrentState() == ConnectorState.STOPPING) {
+ logger.info("{} was requested to stop while it was starting so
will now be stopped", this);
+ stopComponent(scheduler, new CompletableFuture<>());
+ }
}
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 22a148bb709..767c20a5abc 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
@@ -85,6 +85,8 @@ import static org.mockito.Mockito.when;
public class TestStandardConnectorNode {
+ private static final long STOP_NOT_EXPECTED_MILLIS = 250L;
+
private FlowEngine scheduler;
@Mock
@@ -98,11 +100,13 @@ public class TestStandardConnectorNode {
private FlowContextFactory flowContextFactory;
private StateManagerProvider stateManagerProvider;
+ private StartBlockingConnector startBlockingConnector;
@BeforeEach
public void setUp() {
MockitoAnnotations.openMocks(this);
- scheduler = new FlowEngine(1, "flow-engine");
+ // Multiple Threads configured to support concurrent lifecycle
operations
+ scheduler = new FlowEngine(2, "flow-engine");
stateManagerProvider = new MockStateManagerProvider();
when(managedProcessGroup.purge()).thenReturn(CompletableFuture.completedFuture(null));
@@ -135,6 +139,10 @@ public class TestStandardConnectorNode {
@AfterEach
public void teardown() {
+ if (startBlockingConnector != null) {
+ startBlockingConnector.releaseStart();
+ }
+
if (scheduler != null) {
scheduler.close();
}
@@ -297,6 +305,63 @@ public class TestStandardConnectorNode {
assertTrue(startFuture.isDone());
}
+ @Test
+ @Timeout(value = 10, unit = TimeUnit.SECONDS)
+ public void testStopWhileStartingStopsConnectorOnceStartCompletes() throws
Exception {
+ final StartBlockingConnector connector =
createStartBlockingConnector();
+ final StandardConnectorNode connectorNode =
createConnectorNode(connector);
+
+ final Future<Void> startFuture = connectorNode.start(scheduler);
+ assertTrue(connector.awaitStartEntered(5, TimeUnit.SECONDS));
+ assertEquals(ConnectorState.STARTING, connectorNode.getCurrentState());
+
+ final Future<Void> stopFuture = connectorNode.stop(scheduler);
+ assertEquals(ConnectorState.STOPPING, connectorNode.getCurrentState());
+ assertEquals(ConnectorState.STOPPED, connectorNode.getDesiredState());
+
+ // The Connector must not be stopped while its start is still in
progress. The stop is carried out by the
+ // thread performing the start, once that start has finished, so the
Connector remains blocked in start and
+ // in the STOPPING state for as long as the start is held.
+ assertFalse(connector.awaitStopEntered(STOP_NOT_EXPECTED_MILLIS,
TimeUnit.MILLISECONDS));
+ assertEquals(ConnectorState.STOPPING, connectorNode.getCurrentState());
+
+ connector.releaseStart();
+
+ stopFuture.get(5, TimeUnit.SECONDS);
+ startFuture.get(5, TimeUnit.SECONDS);
+
+ // A start that finishes after a stop has been requested must not
leave the Connector reporting RUNNING.
+ assertFalse(connector.wasStoppedWhileStarting());
+ assertEquals(ConnectorState.STOPPED, connectorNode.getCurrentState());
+ assertEquals(ConnectorState.STOPPED, connectorNode.getDesiredState());
+ }
+
+ @Test
+ @Timeout(value = 10, unit = TimeUnit.SECONDS)
+ public void
testStartWhileStopIsPendingForInFlightStartLeavesConnectorRunning() throws
Exception {
+ final StartBlockingConnector connector =
createStartBlockingConnector();
+ final StandardConnectorNode connectorNode =
createConnectorNode(connector);
+
+ connectorNode.start(scheduler);
+ assertTrue(connector.awaitStartEntered(5, TimeUnit.SECONDS));
+
+ final Future<Void> stopFuture = connectorNode.stop(scheduler);
+ assertEquals(ConnectorState.STOPPING, connectorNode.getCurrentState());
+
+ // The desired state returns to RUNNING before the deferred stop has
been carried out, so the Connector must be
+ // stopped and then started again rather than being left in STOPPING.
+ final Future<Void> restartFuture = connectorNode.start(scheduler);
+ assertEquals(ConnectorState.RUNNING, connectorNode.getDesiredState());
+
+ connector.releaseStart();
+
+ stopFuture.get(5, TimeUnit.SECONDS);
+ restartFuture.get(5, TimeUnit.SECONDS);
+
+ assertEquals(ConnectorState.RUNNING, connectorNode.getCurrentState());
+ assertEquals(ConnectorState.RUNNING, connectorNode.getDesiredState());
+ }
+
@Test
public void testCannotDeleteWhenStarting() throws Exception {
// Use a slow-starting connector to test deletion during STARTING state
@@ -1287,6 +1352,11 @@ public class TestStandardConnectorNode {
return createConnectorNode(sleepingConnector);
}
+ private StartBlockingConnector createStartBlockingConnector() {
+ startBlockingConnector = new StartBlockingConnector();
+ return startBlockingConnector;
+ }
+
private StandardConnectorNode createConnectorNode(final Connector
connector) throws FlowUpdateException {
final SecretsManager defaultSecretsManager =
mock(SecretsManager.class);
when(defaultSecretsManager.getAllSecrets()).thenReturn(List.of());
@@ -1966,6 +2036,95 @@ public class TestStandardConnectorNode {
}
}
+ /**
+ * Test connector whose start blocks until it is explicitly released, and
which records whether its stop was
+ * invoked while a start was still in progress. Used to exercise the
interleaving of a stop request with an
+ * in-flight start.
+ */
+ private static class StartBlockingConnector extends AbstractConnector {
+ private static final long MAX_START_BLOCK_SECONDS = 30L;
+
+ private final CountDownLatch startEnteredLatch = new CountDownLatch(1);
+ private final CountDownLatch startReleaseLatch = new CountDownLatch(1);
+ private final CountDownLatch stopEnteredLatch = new CountDownLatch(1);
+ private volatile boolean starting = false;
+ private volatile boolean stoppedWhileStarting = false;
+
+ @Override
+ public VersionedExternalFlow getInitialFlow() {
+ return null;
+ }
+
+ @Override
+ public VersionedExternalFlow getActiveFlow(final FlowContext
activeFlowContext) {
+ return getInitialFlow();
+ }
+
+ @Override
+ public void prepareForUpdate(final FlowContext workingContext, final
FlowContext activeContext) {
+ }
+
+ @Override
+ public List<ConfigurationStep> getConfigurationSteps() {
+ return List.of();
+ }
+
+ @Override
+ public void applyUpdate(final FlowContext workingContext, final
FlowContext activeContext) {
+ }
+
+ @Override
+ protected void onStepConfigured(final String stepName, final
FlowContext workingContext) {
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verifyConfigurationStep(final
String stepName, final Map<String, String> overrides, final FlowContext
flowContext) {
+ return List.of();
+ }
+
+ @Override
+ public void start(final FlowContext activeContext) throws
FlowUpdateException {
+ starting = true;
+ startEnteredLatch.countDown();
+
+ // The wait is bounded so that a test which fails before releasing
the start cannot leave a scheduler
+ // thread blocked indefinitely, which would hang shutdown of the
scheduler during teardown.
+ try {
+ startReleaseLatch.await(MAX_START_BLOCK_SECONDS,
TimeUnit.SECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlowUpdateException(e);
+ } finally {
+ starting = false;
+ }
+ }
+
+ @Override
+ public void stop(final FlowContext activeContext) {
+ if (starting) {
+ stoppedWhileStarting = true;
+ }
+
+ stopEnteredLatch.countDown();
+ }
+
+ public boolean awaitStartEntered(final long timeout, final TimeUnit
unit) throws InterruptedException {
+ return startEnteredLatch.await(timeout, unit);
+ }
+
+ public boolean awaitStopEntered(final long timeout, final TimeUnit
unit) throws InterruptedException {
+ return stopEnteredLatch.await(timeout, unit);
+ }
+
+ public void releaseStart() {
+ startReleaseLatch.countDown();
+ }
+
+ public boolean wasStoppedWhileStarting() {
+ return stoppedWhileStarting;
+ }
+ }
+
/**
* Test connector that allows control over when drainFlowFiles completes
via a CompletableFuture
*/
