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 a5d5e509bf6 NIFI-16288 Allow Connectors to stop while components are
starting or enabling (#11664)
a5d5e509bf6 is described below
commit a5d5e509bf62d452cf90fd9c5cfc4040689bab95
Author: Mark Payne <[email protected]>
AuthorDate: Thu Sep 10 14:27:21 2026 -0400
NIFI-16288 Allow Connectors to stop while components are starting or
enabling (#11664)
---
.../nifi/controller/StandardProcessorNode.java | 80 +++++--
.../controller/service/ServiceStateTransition.java | 53 ++++-
.../service/StandardControllerServiceNode.java | 237 +++++++++++---------
.../connector/StandardConnectorNode.java | 128 ++++++++---
.../StandaloneProcessGroupLifecycle.java | 2 +-
.../scheduling/StandardProcessScheduler.java | 12 +-
.../ControllerServiceEnablingConnector.java | 113 ++++++++++
.../connector/ProcessorStartFailureConnector.java | 95 ++++++++
.../connector/StandardConnectorNodeIT.java | 247 ++++++++++++++++++++-
.../connector/processors/CreateDummyFlowFile.java | 2 +-
.../impl/BlockingEnablingCounterService.java | 122 ++++++++++
.../impl/FailingEnablingCounterService.java | 50 +++++
.../scheduling/TestStandardProcessScheduler.java | 91 ++++++++
.../processors/FailOnScheduledProcessor.java | 36 ++-
.../TestStandardControllerServiceProvider.java | 54 ++++-
.../org.apache.nifi.components.connector.Connector | 2 +
.../org.apache.nifi.controller.ControllerService | 4 +-
.../services/org.apache.nifi.processor.Processor | 3 +-
18 files changed, 1140 insertions(+), 191 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
index e24af1afea7..2c042fd816e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
@@ -187,6 +187,8 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
private SchedulingStrategy schedulingStrategy; // guarded by synchronized
keyword
private ExecutionNode executionNode;
private final Map<Thread, ActiveTask> activeThreads = new
ConcurrentHashMap<>(48);
+ private final AtomicReference<AtomicLong> activeStartupAttempt = new
AtomicReference<>();
+ private final AtomicReference<AtomicLong> runningStartupAttempt = new
AtomicReference<>();
private final int hashCode;
private volatile boolean hasActiveThreads = false;
@@ -1519,7 +1521,9 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
}
if (starting) { // will ensure that the Processor represented by this
node can only be started once
- initiateStart(taskScheduler, administrativeYieldMillis,
timeoutMillis, new AtomicLong(0), processContextFactory,
schedulingAgentCallback, triggerLifecycleMethods);
+ final AtomicLong startupAttemptCount = new AtomicLong(0);
+ activeStartupAttempt.set(startupAttemptCount);
+ initiateStart(taskScheduler, administrativeYieldMillis,
timeoutMillis, startupAttemptCount, processContextFactory,
schedulingAgentCallback, triggerLifecycleMethods);
} else {
final String procName =
processorRef.get().getProcessor().toString();
procLog.warn("Cannot start {} because it is not currently stopped.
Current state is {}", procName, currentState);
@@ -1681,6 +1685,10 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
final AtomicLong startupAttemptCount, final
Supplier<ProcessContext> processContextFactory, final SchedulingAgentCallback
schedulingAgentCallback,
final boolean triggerLifecycleMethods) {
+ if (activeStartupAttempt.get() != startupAttemptCount) {
+ return;
+ }
+
final Processor processor = getProcessor();
final ComponentLog procLog = new
StandardComponentLog(StandardProcessorNode.this.getIdentifier(), processor, new
StandardLoggingContext(StandardProcessorNode.this));
@@ -1692,10 +1700,16 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
// Create a task to invoke the @OnScheduled annotation of the processor
final Callable<Void> startupTask = () -> {
+ if (activeStartupAttempt.get() != startupAttemptCount) {
+ schedulingAgentCallback.onTaskComplete();
+ return null;
+ }
+
final ScheduledState currentScheduleState = scheduledState.get();
if (currentScheduleState == ScheduledState.STOPPING ||
currentScheduleState == ScheduledState.STOPPED || getDesiredState() ==
ScheduledState.STOPPED) {
LOG.info("Aborting start of {}: scheduledState={},
desiredState={}, validationStatus={}",
StandardProcessorNode.this, currentScheduleState,
getDesiredState(), getValidationStatus());
+ activeStartupAttempt.compareAndSet(startupAttemptCount, null);
schedulingAgentCallback.onTaskComplete();
completeStopAction();
return null;
@@ -1707,6 +1721,7 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
final ValidationState validationState =
getValidationState();
procLog.warn("Cannot run once {} because Processor is not
valid (Validation State is {}: {}). Returning to stopped.",
StandardProcessorNode.this, validationState,
validationState.getValidationErrors());
+ activeStartupAttempt.compareAndSet(startupAttemptCount,
null);
schedulingAgentCallback.onTaskComplete();
completeStopAction();
return null;
@@ -1725,11 +1740,11 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
LOG.debug("Cannot start {} because Processor is currently
not valid; will try again after 500 ms", StandardProcessorNode.this);
}
- // re-initiate the entire process
- final Runnable initiateStartTask = () ->
initiateStart(taskScheduler, administrativeYieldMillis, timeoutMillis,
startupAttemptCount,
- processContextFactory, schedulingAgentCallback,
triggerLifecycleMethods);
-
- taskScheduler.schedule(initiateStartTask, 500,
TimeUnit.MILLISECONDS);
+ if (activeStartupAttempt.get() == startupAttemptCount) {
+ final Runnable initiateStartTask = () ->
initiateStart(taskScheduler, administrativeYieldMillis, timeoutMillis,
startupAttemptCount,
+ processContextFactory, schedulingAgentCallback,
triggerLifecycleMethods);
+ taskScheduler.schedule(initiateStartTask, 500,
TimeUnit.MILLISECONDS);
+ }
schedulingAgentCallback.onTaskComplete();
return null;
@@ -1739,14 +1754,32 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
completionTimestampRef.set(System.currentTimeMillis() +
timeoutMillis);
final ProcessContext processContext = processContextFactory.get();
+ final boolean startPermitted;
+ synchronized (this) {
+ final ScheduledState currentState = scheduledState.get();
+ startPermitted = activeStartupAttempt.get() ==
startupAttemptCount && currentState != ScheduledState.STOPPING
+ && currentState != ScheduledState.STOPPED &&
getDesiredState() != ScheduledState.STOPPED;
+ if (startPermitted) {
+ runningStartupAttempt.set(startupAttemptCount);
+ if (triggerLifecycleMethods) {
+ activateThread();
+ }
+ }
+ }
+ if (!startPermitted) {
+ schedulingAgentCallback.onTaskComplete();
+ completeStopAction();
+ return null;
+ }
+
+ boolean startupCompleted = false;
try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(getExtensionManager(),
processor.getClass(), processor.getIdentifier())) {
try {
hasActiveThreads = true;
if (triggerLifecycleMethods) {
LOG.debug("Invoking @OnScheduled methods of {}",
processor);
- activateThread();
try {
ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor,
processContext);
} finally {
@@ -1760,7 +1793,9 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
(desiredState == ScheduledState.RUNNING &&
scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING))
|| (desiredState == ScheduledState.RUN_ONCE &&
scheduledState.compareAndSet(ScheduledState.RUN_ONCE, ScheduledState.RUN_ONCE))
) {
+ startupCompleted = true;
LOG.debug("Successfully completed the @OnScheduled
methods of {}; will now start triggering processor to run", processor);
+
activeStartupAttempt.compareAndSet(startupAttemptCount, null);
schedulingAgentCallback.trigger(); // callback
provided by StandardProcessScheduler to essentially initiate component's
onTrigger() cycle
} else {
LOG.info("Successfully invoked @OnScheduled methods of
{} but scheduled state is no longer STARTING so will stop processor now;
current state = {}, desired state = {}",
@@ -1792,7 +1827,7 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
} finally {
schedulingAgentCallback.onTaskComplete();
}
- } catch (Exception e) {
+ } catch (final Exception e) {
final Throwable cause = (e instanceof
InvocationTargetException) ? e.getCause() : e;
procLog.error("Failed to properly initialize Processor. If
still scheduled to run, NiFi will attempt to "
+ "initialize and run the Processor again after the
'Administrative Yield Duration' has elapsed. Failure is due to " + cause,
cause);
@@ -1810,13 +1845,17 @@ public class StandardProcessorNode extends
ProcessorNode implements Connectable
}
// make sure we only continue retry loop if STOP action wasn't
initiated
- if (scheduledState.get() != ScheduledState.STOPPING &&
scheduledState.get() != ScheduledState.RUN_ONCE) {
- // re-initiate the entire process
+ if (activeStartupAttempt.get() == startupAttemptCount &&
scheduledState.get() == ScheduledState.STARTING && getDesiredState() ==
ScheduledState.RUNNING) {
final Runnable initiateStartTask = () ->
initiateStart(taskScheduler, administrativeYieldMillis, timeoutMillis,
startupAttemptCount,
processContextFactory, schedulingAgentCallback,
triggerLifecycleMethods);
-
taskScheduler.schedule(initiateStartTask,
administrativeYieldMillis, TimeUnit.MILLISECONDS);
} else {
+ activeStartupAttempt.compareAndSet(startupAttemptCount,
null);
+ completeStopAction();
+ }
+ } finally {
+ runningStartupAttempt.compareAndSet(startupAttemptCount, null);
+ if (!startupCompleted && activeStartupAttempt.get() !=
startupAttemptCount && getDesiredState() == ScheduledState.STOPPED) {
completeStopAction();
}
}
@@ -1828,7 +1867,8 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
try {
// Trigger the task in a background thread.
taskFuture = schedulingAgentCallback.scheduleTask(startupTask);
- } catch (RejectedExecutionException rejectedExecutionException) {
+ } catch (final RejectedExecutionException rejectedExecutionException) {
+ activeStartupAttempt.compareAndSet(startupAttemptCount, null);
final ValidationState validationState = getValidationState();
LOG.error("Unable to start {}. Last known validation state was {}
: {}", this, validationState, validationState.getValidationErrors(),
rejectedExecutionException);
return;
@@ -1993,9 +2033,19 @@ public class StandardProcessorNode extends ProcessorNode
implements Connectable
// before stop() was called. If that happens the stop processor
// routine will be initiated in start() method, otherwise the IF
// part will handle the stop processor routine.
- final boolean updated =
this.scheduledState.compareAndSet(ScheduledState.STARTING,
ScheduledState.STOPPING);
- if (updated) {
- LOG.debug("Transitioned state of {} from STARTING to
STOPPING", this);
+ synchronized (this) {
+ final boolean updated =
this.scheduledState.compareAndSet(ScheduledState.STARTING,
ScheduledState.STOPPING);
+ if (updated) {
+ LOG.debug("Transitioned state of {} from STARTING to
STOPPING", this);
+ activeStartupAttempt.set(null);
+ for (final Thread activeThread : activeThreads.keySet()) {
+ activeThread.interrupt();
+ }
+
+ if (activeThreads.isEmpty() && runningStartupAttempt.get()
== null) {
+ completeStopAction();
+ }
+ }
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
index 67de046c2cc..734b16731c4 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/ServiceStateTransition.java
@@ -25,6 +25,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
@@ -67,7 +68,7 @@ public class ServiceStateTransition {
}
}
- public boolean enable(final ControllerServiceReference
controllerServiceReference) {
+ public boolean enable(final ControllerServiceReference
controllerServiceReference, final CompletableFuture<?> enabledFuture) {
writeLock.lock();
try {
if (state == ControllerServiceState.ENABLED) {
@@ -75,8 +76,8 @@ public class ServiceStateTransition {
return true;
}
- if (state != ControllerServiceState.ENABLING) {
- logger.debug("{} cannot be transitioned to enabled because
it's not currently ENABLING but rather {}", controllerServiceNode, state);
+ if (state != ControllerServiceState.ENABLING ||
!enabledFutures.contains(enabledFuture)) {
+ logger.debug("{} cannot be transitioned to enabled because the
enable request is no longer active and the current state is {}",
controllerServiceNode, state);
return false;
}
@@ -84,6 +85,7 @@ public class ServiceStateTransition {
logger.debug("{} is now fully ENABLED", controllerServiceNode);
enabledFutures.forEach(future -> future.complete(null));
+ enabledFutures.clear();
} finally {
writeLock.unlock();
}
@@ -108,31 +110,58 @@ public class ServiceStateTransition {
return true;
}
- public boolean transitionToDisabling(final ControllerServiceState
expectedState, final CompletableFuture<?> disabledFuture) {
+ public boolean isEnabling(final CompletableFuture<?> enabledFuture) {
+ readLock.lock();
+ try {
+ return state == ControllerServiceState.ENABLING &&
enabledFutures.contains(enabledFuture);
+ } finally {
+ readLock.unlock();
+ }
+ }
+
+ public ControllerServiceState transitionToDisabling(final
CompletableFuture<?> disabledFuture) {
writeLock.lock();
try {
- if (expectedState != state) {
- logger.debug("{} cannot be transitioned to DISABLING because
its state is {}, not the expected {}", controllerServiceNode, state,
expectedState);
- return false;
+ final ControllerServiceState previousState = state;
+ if (previousState == ControllerServiceState.DISABLED) {
+ disabledFuture.complete(null);
+ } else if (previousState == ControllerServiceState.DISABLING) {
+ disabledFutures.add(disabledFuture);
+ } else {
+ state = ControllerServiceState.DISABLING;
+ stateChangeCondition.signalAll();
+ disabledFutures.add(disabledFuture);
}
- state = ControllerServiceState.DISABLING;
- stateChangeCondition.signalAll();
- disabledFutures.add(disabledFuture);
- return true;
+ if (previousState == ControllerServiceState.ENABLING) {
+ for (final CompletableFuture<?> enabledFuture :
enabledFutures) {
+ enabledFuture.completeExceptionally(new
CancellationException("Controller Service enablement cancelled by disable
request"));
+ }
+
+ enabledFutures.clear();
+ }
+
+ return previousState;
} finally {
writeLock.unlock();
}
}
- public void disable() {
+ public boolean disable() {
writeLock.lock();
try {
+ if (state != ControllerServiceState.DISABLING) {
+ logger.debug("{} cannot be transitioned to DISABLED because
its state is {}", controllerServiceNode, state);
+ return false;
+ }
+
state = ControllerServiceState.DISABLED;
logger.info("{} is now fully DISABLED", controllerServiceNode);
stateChangeCondition.signalAll();
disabledFutures.forEach(future -> future.complete(null));
+ disabledFutures.clear();
+ return true;
} finally {
writeLock.unlock();
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
index b2010072027..a2fb24a7c5e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java
@@ -94,7 +94,9 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -132,6 +134,8 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
private final VerifiableComponentFactory verifiableComponentFactory;
private final AtomicBoolean active;
+ private final AtomicBoolean enablingTaskRunning = new AtomicBoolean();
+ private final AtomicReference<Future<?>> enablingTask = new
AtomicReference<>();
public StandardControllerServiceNode(final
LoggableComponent<ControllerService> implementation, final
LoggableComponent<ControllerService> proxiedControllerService,
final
ControllerServiceInvocationHandler invocationHandler, final String id, final
ValidationContextFactory validationContextFactory,
@@ -629,9 +633,9 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
* Upon successful invocation of @OnEnabled this service will be
transitioned to
* ENABLED state.
* <br>
- * In the event where enabling took longer then expected by the user and
such user
- * initiated disable operation, this service will be automatically
disabled as soon
- * as it reached ENABLED state.
+ * When a disable operation is initiated during enabling, the enable
future is completed exceptionally
+ * with a {@link CancellationException}. The service remains DISABLING
until the enable lifecycle method returns
+ * and the corresponding disable lifecycle method finishes.
*/
@Override
public CompletableFuture<Void> enable(final ScheduledExecutorService
scheduler, final long administrativeYieldMillis, final boolean
completeExceptionallyOnFailure) {
@@ -644,130 +648,154 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
final CompletableFuture<Void> future = new CompletableFuture<>();
- if
(!stateTransition.transitionToEnabling(ControllerServiceState.DISABLED,
future)) {
- future.complete(null);
- return future;
+ final boolean transitionedToEnabling;
+ synchronized (active) {
+ transitionedToEnabling =
stateTransition.transitionToEnabling(ControllerServiceState.DISABLED, future);
+ if (transitionedToEnabling) {
+ active.set(true);
+ }
}
- synchronized (active) {
- this.active.set(true);
+ if (!transitionedToEnabling) {
+ future.complete(null);
+ return future;
}
final AtomicLong enablingDelay = new AtomicLong(0);
final AtomicLong validationDelay = new AtomicLong(0);
final ControllerServiceProvider controllerServiceProvider =
this.serviceProvider;
final StandardControllerServiceNode serviceNode = this;
- scheduler.execute(new Runnable() {
+ final Runnable enablingRunnable = new Runnable() {
@Override
public void run() {
final ConfigurationContext configContext =
providedConfigurationContext == null
? new StandardConfigurationContext(serviceNode,
controllerServiceProvider, null)
: providedConfigurationContext;
+ boolean enabled = false;
+ synchronized (active) {
+ if (!active.get() || !stateTransition.isEnabling(future)) {
+ return;
+ }
- if (!isActive()) {
- LOG.warn("Enabling {} stopped: no active status",
serviceNode);
- stateTransition.disable();
- future.complete(null);
- return;
- }
-
- // Perform validation - if a ConfigurationContext was
provided, validate against its properties
- final ValidationState validationState;
- if (providedConfigurationContext == null) {
- performValidation();
- validationState = getValidationState();
- } else {
- final Map<String, String> properties =
providedConfigurationContext.getAllProperties();
- final ValidationContext validationContext =
createValidationContext(properties, getAnnotationData(), getParameterLookup(),
true);
- validationState = performValidation(validationContext);
+ enablingTaskRunning.set(true);
}
- final ValidationStatus validationStatus =
validationState.getStatus();
- if (validationStatus == ValidationStatus.VALID) {
- LOG.debug("Enabling {} proceeding after performing
validation", serviceNode);
- } else {
- final Collection<ValidationResult> errors =
validationState.getValidationErrors();
- if (completeExceptionallyOnFailure) {
- future.completeExceptionally(new
IllegalStateException("Enabling %s failed: Validation Status [%s] Errors
%s".formatted(serviceNode, validationStatus, errors)));
+ try {
+ // Perform validation - if a ConfigurationContext was
provided, validate against its properties
+ final ValidationState validationState;
+ if (providedConfigurationContext == null) {
+ performValidation();
+ validationState = getValidationState();
+ } else {
+ final Map<String, String> properties =
providedConfigurationContext.getAllProperties();
+ final ValidationContext validationContext =
createValidationContext(properties, getAnnotationData(), getParameterLookup(),
true);
+ validationState = performValidation(validationContext);
}
- final long selectedValidationDelay =
getDelay(validationDelay, INCREMENTAL_VALIDATION_DELAY_MS);
+ final ValidationStatus validationStatus =
validationState.getStatus();
+ if (validationStatus == ValidationStatus.VALID) {
+ LOG.debug("Enabling {} proceeding after performing
validation", serviceNode);
+ } else {
+ final Collection<ValidationResult> errors =
validationState.getValidationErrors();
+ if (completeExceptionallyOnFailure) {
+ future.completeExceptionally(new
IllegalStateException("Enabling %s failed: Validation Status [%s] Errors
%s".formatted(serviceNode, validationStatus, errors)));
+ }
- // Log warning on repeated validation rescheduling
- if (selectedValidationDelay > MAXIMUM_DELAY.toMillis()) {
- LOG.warn("Validation rescheduled in {} ms for {}
Errors {}", selectedValidationDelay, serviceNode, errors);
- }
+ final long selectedValidationDelay =
getDelay(validationDelay, INCREMENTAL_VALIDATION_DELAY_MS);
- try {
- scheduler.schedule(this, selectedValidationDelay,
TimeUnit.MILLISECONDS);
- LOG.debug("Validation rescheduled in {} ms for {}",
selectedValidationDelay, serviceNode);
- } catch (final RejectedExecutionException e) {
- LOG.debug("Validation rescheduling rejected for {}",
serviceNode, e);
- future.completeExceptionally(new
IllegalStateException("Enabling %s rejected: Validation Status [%s] Errors
%s".formatted(serviceNode, validationStatus, errors)));
+ // Log warning on repeated validation rescheduling
+ if (selectedValidationDelay >
MAXIMUM_DELAY.toMillis()) {
+ LOG.warn("Validation rescheduled in {} ms for {}
Errors {}", selectedValidationDelay, serviceNode, errors);
+ }
+
+ try {
+ if (scheduleEnableTask(scheduler, this,
selectedValidationDelay, future)) {
+ LOG.debug("Validation rescheduled in {} ms for
{}", selectedValidationDelay, serviceNode);
+ }
+ } catch (final RejectedExecutionException e) {
+ LOG.debug("Validation rescheduling rejected for
{}", serviceNode, e);
+ future.completeExceptionally(new
IllegalStateException(
+ "Enabling %s rejected: Validation Status [%s]
Errors %s".formatted(serviceNode, validationStatus, errors)));
+ }
+
+ return;
}
- // Enable command rescheduled or rejected
- return;
- }
+ synchronized (active) {
+ if (!active.get() ||
!stateTransition.isEnabling(future)) {
+ return;
+ }
+ }
- final ControllerService controllerService =
getControllerServiceImplementation();
- try {
+ final ControllerService controllerService =
getControllerServiceImplementation();
try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(getExtensionManager(),
controllerService.getClass(), getIdentifier())) {
ReflectionUtils.invokeMethodsWithAnnotation(OnEnabled.class, controllerService,
configContext);
}
boolean shouldEnable;
synchronized (active) {
- shouldEnable = active.get() &&
stateTransition.enable(getReferences()); // Transitioning the state to ENABLED
will complete our future.
+ shouldEnable = active.get() &&
stateTransition.enable(getReferences(), future);
}
- if (!shouldEnable) {
- LOG.info("Disabling {} after enabled due to disable
action initiated", serviceNode);
- // Can only happen if user initiated DISABLE operation
before service finished enabling. It's state will be
- // set to DISABLING (see disable() operation)
- invokeDisable(configContext);
- stateTransition.disable();
- future.complete(null);
- } else {
+ if (shouldEnable) {
+ enabled = true;
LOG.info("Enabled {}", serviceNode);
+ } else {
+ LOG.info("Disabling {} after enable lifecycle
completed while a disable action was in progress", serviceNode);
+ invokeDisable(configContext);
}
} catch (final Exception e) {
if (completeExceptionallyOnFailure) {
future.completeExceptionally(e);
}
+ final ControllerService controllerService =
getControllerServiceImplementation();
final Throwable cause = e instanceof
InvocationTargetException ? e.getCause() : e;
final ComponentLog componentLog = new
StandardComponentLog(getIdentifier(), controllerService, new
StandardLoggingContext(serviceNode));
componentLog.error("Failed to invoke @OnEnabled method",
cause);
+
invokeDisable(configContext);
- if (isActive()) {
+ if (isActive() && stateTransition.isEnabling(future)) {
// Increment enabling delay to avoid excessive retries
final long selectedEnablingDelay =
getDelay(enablingDelay, administrativeYieldMillis);
- scheduler.schedule(this, selectedEnablingDelay,
TimeUnit.MILLISECONDS);
- } else {
- stateTransition.disable();
+ scheduleEnableTask(scheduler, this,
selectedEnablingDelay, future);
+ }
+ } finally {
+ final boolean completeDisable;
+ synchronized (active) {
+ enablingTaskRunning.set(false);
+ completeDisable = !enabled && !active.get() &&
stateTransition.getState() == ControllerServiceState.DISABLING;
+ }
+
+ if (completeDisable) {
+ completeDisabling();
}
}
}
- });
+ };
+ scheduleEnableTask(scheduler, enablingRunnable, 0, future);
return future;
}
+ private boolean scheduleEnableTask(final ScheduledExecutorService
scheduler, final Runnable task, final long delay,
+ final CompletableFuture<?> enabledFuture) {
+ synchronized (active) {
+ if (!active.get() || !stateTransition.isEnabling(enabledFuture)) {
+ return false;
+ }
+
+ enablingTask.set(scheduler.schedule(task, delay,
TimeUnit.MILLISECONDS));
+ return true;
+ }
+ }
+
/**
- * Will atomically disable this service by invoking its @OnDisabled
operation.
- * It uses CAS operation on {@link #stateTransition} to transition this
service
- * from ENABLED to DISABLING state. If such transition succeeds the service
- * will be de-activated (see {@link ControllerServiceNode#isActive()}).
- * If such transition doesn't succeed (the service is still in ENABLING
state)
- * then the service will still be transitioned to DISABLING state to
ensure that
- * no other transition could happen on this service. However in such event
- * (e.g., its @OnEnabled finally succeeded), the {@link
#enable(ScheduledExecutorService, long, boolean)}
- * operation will initiate service disabling javadoc for (see {@link
#enable(ScheduledExecutorService, long, boolean)}
- * <br>
- * Upon successful invocation of @OnDisabled this service will be
transitioned to
- * DISABLED state.
+ * Atomically transitions this service to DISABLING. Services that are
ENABLING complete pending enable futures exceptionally
+ * with a {@link CancellationException} and request interruption of the
active enable task.
+ * Services that are ENABLED invoke their @OnDisabled methods before
transitioning to DISABLED.
+ * All callers receive a future that completes when the service is
DISABLED.
*/
@Override
public CompletableFuture<Void> disable(final ScheduledExecutorService
scheduler) {
@@ -777,47 +805,36 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
* service since it will attempt to transition service state from
* ENABLING to ENABLED but only if it's active.
*/
- synchronized (this.active) {
- this.active.set(false);
- }
-
final CompletableFuture<Void> future = new CompletableFuture<>();
- // If already disabled, complete immediately
- if (getState() == ControllerServiceState.DISABLED) {
- future.complete(null);
- return future;
+ final ControllerServiceState previousState;
+ final Future<?> task;
+ final boolean taskRunning;
+ synchronized (active) {
+ active.set(false);
+ previousState = stateTransition.transitionToDisabling(future);
+ task = enablingTask.getAndSet(null);
+ taskRunning = enablingTaskRunning.get();
}
- final boolean transitioned =
this.stateTransition.transitionToDisabling(ControllerServiceState.ENABLING,
future);
- if (transitioned) {
- // If we transitioned from ENABLING to DISABLING, we need to
immediately complete the disable
- // because the enable task may be scheduled to run far in the
future (up to 10 minutes) due to
- // validation retries. Rather than making the user wait, we
immediately transition to DISABLED.
- scheduler.execute(() -> {
- stateTransition.disable();
+ if (previousState == ControllerServiceState.ENABLING) {
+ if (task != null) {
+ task.cancel(true);
+ }
+
+ if (!taskRunning) {
+ scheduler.execute(this::completeDisabling);
+ }
- // Now all components that reference this service will be
invalid. Trigger validation to occur so that
- // this is reflected in any response that may go back to a
user/client.
- for (final ComponentNode component :
getReferences().getReferencingComponents()) {
- component.performValidation();
- }
- });
return future;
}
- if
(this.stateTransition.transitionToDisabling(ControllerServiceState.ENABLED,
future)) {
+ if (previousState == ControllerServiceState.ENABLED) {
final ConfigurationContext configContext = new
StandardConfigurationContext(this, this.serviceProvider, null);
scheduler.execute(() -> {
try {
invokeDisable(configContext);
} finally {
- stateTransition.disable();
-
- // Now all components that reference this service will be
invalid. Trigger validation to occur so that
- // this is reflected in any response that may go back to a
user/client.
- for (final ComponentNode component :
getReferences().getReferencingComponents()) {
- component.performValidation();
- }
+ completeDisabling();
}
});
}
@@ -825,12 +842,22 @@ public class StandardControllerServiceNode extends
AbstractComponentNode impleme
return future;
}
- private void invokeDisable(ConfigurationContext configContext) {
+ private void completeDisabling() {
+ if (!stateTransition.disable()) {
+ return;
+ }
+
+ for (final ComponentNode component :
getReferences().getReferencingComponents()) {
+ component.performValidation();
+ }
+ }
+
+ private void invokeDisable(final ConfigurationContext configContext) {
final ControllerService controllerService =
getControllerServiceImplementation();
try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(getExtensionManager(),
controllerService.getClass(), getIdentifier())) {
ReflectionUtils.invokeMethodsWithAnnotation(OnDisabled.class,
controllerService, configContext);
LOG.debug("Successfully disabled {}", this);
- } catch (Exception e) {
+ } catch (final Exception e) {
final Throwable cause = e instanceof InvocationTargetException ?
e.getCause() : e;
final ComponentLog componentLog = new
StandardComponentLog(getIdentifier(), controllerService, new
StandardLoggingContext(StandardControllerServiceNode.this));
componentLog.error("Failed to invoke @OnDisabled method due to
{}", cause);
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 5d3f7c582dc..d5d836dfc64 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
@@ -90,6 +90,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -127,6 +128,8 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
private volatile String name;
private volatile FrameworkConnectorInitializationContext
initializationContext;
+ // Serializes Connector start and stop invocations so that a stop request
cannot complete before an in-flight start invocation returns.
+ private final ReentrantLock componentLifecycleLock = new ReentrantLock();
private final Object loggingAttributesLock = new Object();
private volatile Map<String, String> customLoggingAttributes = Map.of();
private volatile Map<String, String> mergedLoggingAttributes = Map.of();
@@ -861,8 +864,8 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
// 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);
+ stopManagedProcessGroup(scheduler);
return stopCompleteFuture;
}
}
@@ -872,6 +875,33 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
return stopCompleteFuture;
}
+ private void stopManagedProcessGroup(final FlowEngine scheduler) {
+ if (getCurrentState() != ConnectorState.STOPPING) {
+ return;
+ }
+
+ logger.info("Stopping the managed Process Group for {} so that its
startup can finish and the Connector can stop", this);
+ final CompletableFuture<Void> processGroupStopFuture;
+ try {
+ processGroupStopFuture =
activeFlowContext.getRootGroup().getLifecycle().stop();
+ } catch (final Exception e) {
+ logger.warn("Failed to stop the managed Process Group for {}. The
Connector cannot finish stopping until its components stop, so this will be
tried again in 10 seconds", this, e);
+ scheduler.schedule(() -> stopManagedProcessGroup(scheduler), 10,
TimeUnit.SECONDS);
+ return;
+ }
+
+ processGroupStopFuture.whenComplete((result, failure) -> {
+ if (failure != null) {
+ logger.warn("Failed to stop the managed Process Group for {}.
The Connector cannot finish stopping until its components stop, so this will be
tried again in 10 seconds",
+ this, failure);
+ scheduler.schedule(() -> stopManagedProcessGroup(scheduler),
10, TimeUnit.SECONDS);
+ return;
+ }
+
+ scheduler.schedule(() -> completeDeferredStop(scheduler), 0,
TimeUnit.SECONDS);
+ });
+ }
+
@Override
public Future<Void> drainFlowFiles() {
logger.debug("Draining FlowFiles for {}", this);
@@ -990,49 +1020,64 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
}
private void stopComponent(final FlowEngine scheduler, final
CompletableFuture<Void> stopCompleteFuture) {
- logger.debug("Stopping component for {}", this);
- try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(extensionManager,
connectorDetails.getConnector().getClass(), getIdentifier())) {
- connectorDetails.getConnector().stop(activeFlowContext);
- } catch (final Exception e) {
- logger.error("Failed to stop {}. Will try again in 10 seconds",
this, e);
- scheduler.schedule(() -> stopComponent(scheduler,
stopCompleteFuture), 10, TimeUnit.SECONDS);
- return;
- }
+ componentLifecycleLock.lock();
+ try {
+ if (getCurrentState() != ConnectorState.STOPPING) {
+ return;
+ }
- stateTransition.setCurrentState(ConnectorState.STOPPED);
- stopCompleteFuture.complete(null);
- logger.info("Successfully stopped {}", this);
+ logger.debug("Stopping component for {}", this);
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(extensionManager,
connectorDetails.getConnector().getClass(), getIdentifier())) {
+ connectorDetails.getConnector().stop(activeFlowContext);
+ } catch (final Exception e) {
+ logger.error("Failed to stop {}. Will try again in 10
seconds", this, e);
+ scheduler.schedule(() -> stopComponent(scheduler,
stopCompleteFuture), 10, TimeUnit.SECONDS);
+ return;
+ }
- final ConnectorState desiredState = getDesiredState();
- if (desiredState == ConnectorState.RUNNING) {
- logger.info("{} was requested to be RUNNING while it was stopping
so will attempt to start again", this);
- start(scheduler, new CompletableFuture<>());
+ stateTransition.setCurrentState(ConnectorState.STOPPED);
+ stopCompleteFuture.complete(null);
+ logger.info("Successfully stopped {}", this);
+
+ final ConnectorState desiredState = getDesiredState();
+ if (desiredState == ConnectorState.RUNNING) {
+ logger.info("{} was requested to be RUNNING while it was
stopping so will attempt to start again", this);
+ start(scheduler, new CompletableFuture<>());
+ }
+ } finally {
+ componentLifecycleLock.unlock();
}
}
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) {
- 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);
+ componentLifecycleLock.lock();
+ try {
+ 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);
- } else {
- logger.error("Failed to start {} retrying in 10 seconds",
this, e);
- scheduler.schedule(() -> startComponent(scheduler,
startCompleteFuture), 10, TimeUnit.SECONDS);
+ return;
}
- return;
+ try (final NarCloseable ignored =
NarCloseable.withComponentNarLoader(extensionManager,
connectorDetails.getConnector().getClass(), getIdentifier())) {
+ connectorDetails.getConnector().start(activeFlowContext);
+ } catch (final Exception e) {
+ 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;
+ }
+ } finally {
+ componentLifecycleLock.unlock();
}
+ // Reconcile the state after releasing the lock so a completed disable
operation can acquire it and finish a pending stop.
// 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.
@@ -1048,12 +1093,21 @@ public class StandardConnectorNode implements
ConnectorNode, GroupedComponent {
}
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<>());
+ if (!componentLifecycleLock.tryLock()) {
+ logger.debug("{} has not finished its start invocation, so the
Connector stop will be checked again", this);
+ scheduler.schedule(() -> completeDeferredStop(scheduler), 100,
TimeUnit.MILLISECONDS);
+ return;
}
- }
+ try {
+ if (getCurrentState() == ConnectorState.STOPPING) {
+ logger.info("{} was requested to stop while it was starting so
will now be stopped", this);
+ stopComponent(scheduler, new CompletableFuture<>());
+ }
+ } finally {
+ componentLifecycleLock.unlock();
+ }
+ }
@Override
public void verifyCanDelete() {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
index 3ee3d4a641e..fcc078f3c74 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/connector/facades/standalone/StandaloneProcessGroupLifecycle.java
@@ -425,7 +425,7 @@ public class StandaloneProcessGroupLifecycle implements
ProcessGroupLifecycle {
final Collection<ProcessorNode> processors = recursive ?
processGroup.findAllProcessors() : processGroup.getProcessors();
final List<CompletableFuture<Void>> stopFutures = new ArrayList<>();
for (final ProcessorNode processor : processors) {
- final ScheduledState processorState =
processor.getScheduledState();
+ final ScheduledState processorState =
processor.getPhysicalScheduledState();
if (processorState == ScheduledState.DISABLED || processorState ==
ScheduledState.STOPPED) {
logger.debug("Not stopping Processor {} because its state is
{}", processor, processorState);
continue;
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
index 58fd1e87b3d..6f8744d728e 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
@@ -71,6 +71,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -104,6 +105,7 @@ public final class StandardProcessScheduler implements
ProcessScheduler {
private final ReloadComponent reloadComponent;
private final ConcurrentMap<SchedulingStrategy, SchedulingAgent>
strategyAgentMap = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, CompletableFuture<Void>>
processorStartFutures = new ConcurrentHashMap<>();
// thread pool for starting/stopping components
private volatile boolean shutdown = false;
@@ -424,6 +426,7 @@ public final class StandardProcessScheduler implements
ProcessScheduler {
getSchedulingAgent(procNode).schedule(procNode,
lifecycleState);
}
+ processorStartFutures.remove(procNode.getIdentifier(), future);
future.complete(null);
}
@@ -443,6 +446,7 @@ public final class StandardProcessScheduler implements
ProcessScheduler {
procNode.reloadAdditionalResourcesIfNecessary();
+ processorStartFutures.put(procNode.getIdentifier(), future);
procNode.start(componentMonitoringThreadPool,
administrativeYieldMillis, processorStartTimeoutMillis, processContextFactory,
callback, failIfStopping, scheduleActions);
return future;
}
@@ -595,7 +599,13 @@ public final class StandardProcessScheduler implements
ProcessScheduler {
getStateManager(procNode), lifecycleState::isTerminated,
nodeTypeProvider);
LOG.info("Stopping {}", procNode);
- return procNode.stop(this, this.componentLifeCycleThreadPool,
processContext, getSchedulingAgent(procNode), lifecycleState, lifecycleMethods);
+ final CompletableFuture<Void> stopFuture = procNode.stop(this,
this.componentLifeCycleThreadPool, processContext,
getSchedulingAgent(procNode), lifecycleState, lifecycleMethods);
+ final CompletableFuture<Void> startFuture =
processorStartFutures.remove(procNode.getIdentifier());
+ if (startFuture != null) {
+ startFuture.completeExceptionally(new
CancellationException("Processor start cancelled by stop request"));
+ }
+
+ return stopFuture;
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ControllerServiceEnablingConnector.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ControllerServiceEnablingConnector.java
new file mode 100644
index 00000000000..5bd166a5a9a
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ControllerServiceEnablingConnector.java
@@ -0,0 +1,113 @@
+/*
+ * 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.components.ConfigVerificationResult;
+import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.components.connector.processors.CreateDummyFlowFile;
+import
org.apache.nifi.components.connector.services.impl.BlockingEnablingCounterService;
+import
org.apache.nifi.components.connector.services.impl.FailingEnablingCounterService;
+import org.apache.nifi.components.connector.util.VersionedFlowUtils;
+import org.apache.nifi.flow.Bundle;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedExternalFlow;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+
+import java.util.List;
+import java.util.Map;
+
+public class ControllerServiceEnablingConnector extends AbstractConnector {
+
+ static final String BLOCKING_ENABLING = "BLOCKING_ENABLING";
+ static final String FAILING_ENABLING = "FAILING_ENABLING";
+ static final String CONFIGURATION_STEP_NAME = "Configuration";
+
+ static final ConnectorPropertyDescriptor ENABLING_BEHAVIOR = new
ConnectorPropertyDescriptor.Builder()
+ .name("Enabling Behavior")
+ .description("How the managed Controller Service should behave while
enabling")
+ .type(PropertyType.STRING)
+ .required(true)
+ .defaultValue(BLOCKING_ENABLING)
+ .allowableValues(BLOCKING_ENABLING, FAILING_ENABLING)
+ .build();
+
+ private static final ConnectorPropertyGroup PROPERTY_GROUP = new
ConnectorPropertyGroup.Builder()
+ .name("Controller Service Settings")
+ .addProperty(ENABLING_BEHAVIOR)
+ .build();
+
+ private static final ConfigurationStep CONFIGURATION_STEP = new
ConfigurationStep.Builder()
+ .name(CONFIGURATION_STEP_NAME)
+ .propertyGroups(List.of(PROPERTY_GROUP))
+ .build();
+
+ @Override
+ public List<ConfigurationStep> getConfigurationSteps() {
+ return List.of(CONFIGURATION_STEP);
+ }
+
+ @Override
+ public VersionedExternalFlow getInitialFlow() {
+ return
VersionedFlowUtils.loadFlowFromResource("flows/generate-duplicate-log-flow.json");
+ }
+
+ @Override
+ public VersionedExternalFlow getActiveFlow(final FlowContext
activeFlowContext) {
+ final String enablingBehavior =
activeFlowContext.getConfigurationContext().getProperty(CONFIGURATION_STEP,
ENABLING_BEHAVIOR).getValue();
+ return buildFlow(enablingBehavior);
+ }
+
+ @Override
+ public void applyUpdate(final FlowContext workingContext, final
FlowContext activeContext) throws FlowUpdateException {
+ getInitializationContext().updateFlow(activeContext,
getActiveFlow(workingContext));
+ }
+
+ @Override
+ public void onStepConfigured(final String stepName, final FlowContext
workingContext) throws FlowUpdateException {
+ getInitializationContext().updateFlow(workingContext,
getActiveFlow(workingContext));
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verifyConfigurationStep(final String
stepName, final Map<String, String> overrides, final FlowContext flowContext) {
+ return List.of();
+ }
+
+ private VersionedExternalFlow buildFlow(final String enablingBehavior) {
+ final VersionedExternalFlow externalFlow =
VersionedFlowUtils.loadFlowFromResource("flows/generate-duplicate-log-flow.json");
+ final VersionedProcessGroup rootGroup = externalFlow.getFlowContents();
+
+ final Bundle systemBundle = new Bundle();
+ systemBundle.setArtifact("system");
+ systemBundle.setGroup("default");
+ systemBundle.setVersion("unversioned");
+
+ final String serviceType = BLOCKING_ENABLING.equals(enablingBehavior)
+ ? BlockingEnablingCounterService.class.getName()
+ : FailingEnablingCounterService.class.getName();
+
+ final VersionedControllerService controllerService =
VersionedFlowUtils.addControllerService(rootGroup, serviceType, systemBundle,
"Managed Service");
+
+ final VersionedProcessor generateProcessor =
VersionedFlowUtils.findProcessor(rootGroup,
+ processor ->
processor.getType().equals(CreateDummyFlowFile.class.getName())).orElseThrow();
+
+ generateProcessor.getProperties().put("Counter Service",
controllerService.getIdentifier());
+
+ return externalFlow;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ProcessorStartFailureConnector.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ProcessorStartFailureConnector.java
new file mode 100644
index 00000000000..85e72e70b2f
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/ProcessorStartFailureConnector.java
@@ -0,0 +1,95 @@
+/*
+ * 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.components.ConfigVerificationResult;
+import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.components.connector.processors.CreateDummyFlowFile;
+import
org.apache.nifi.components.connector.services.impl.StandardCounterService;
+import org.apache.nifi.components.connector.util.VersionedFlowUtils;
+import
org.apache.nifi.controller.scheduling.processors.FailOnScheduledProcessor;
+import org.apache.nifi.flow.Bundle;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedExternalFlow;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+
+import java.util.List;
+import java.util.Map;
+
+public class ProcessorStartFailureConnector extends AbstractConnector {
+
+ @Override
+ public List<ConfigurationStep> getConfigurationSteps() {
+ return List.of();
+ }
+
+ @Override
+ public VersionedExternalFlow getInitialFlow() {
+ return buildFlow();
+ }
+
+ @Override
+ public VersionedExternalFlow getActiveFlow(final FlowContext
activeFlowContext) {
+ return buildFlow();
+ }
+
+ @Override
+ public void applyUpdate(final FlowContext workingContext, final
FlowContext activeContext) throws FlowUpdateException {
+ getInitializationContext().updateFlow(activeContext,
getActiveFlow(workingContext));
+ }
+
+ @Override
+ public void onStepConfigured(final String stepName, final FlowContext
workingContext) throws FlowUpdateException {
+ getInitializationContext().updateFlow(workingContext,
getActiveFlow(workingContext));
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verifyConfigurationStep(final String
stepName, final Map<String, String> overrides, final FlowContext flowContext) {
+ return List.of();
+ }
+
+ private VersionedExternalFlow buildFlow() {
+ final VersionedExternalFlow externalFlow =
VersionedFlowUtils.loadFlowFromResource("flows/generate-duplicate-log-flow.json");
+ final VersionedProcessGroup rootGroup = externalFlow.getFlowContents();
+
+ final Bundle systemBundle = new Bundle();
+ systemBundle.setArtifact("system");
+ systemBundle.setGroup("default");
+ systemBundle.setVersion("unversioned");
+
+ final VersionedControllerService controllerService =
VersionedFlowUtils.addControllerService(
+ rootGroup, StandardCounterService.class.getName(), systemBundle,
"Managed Service");
+ final VersionedProcessor processor =
VersionedFlowUtils.findProcessor(rootGroup,
+ candidate ->
candidate.getType().equals(CreateDummyFlowFile.class.getName())).orElseThrow();
+ processor.setType(FailOnScheduledProcessor.class.getName());
+ processor.setName("Start Failure Processor");
+ processor.getProperties().clear();
+
processor.getProperties().put(FailOnScheduledProcessor.MANAGED_SERVICE.getName(),
controllerService.getIdentifier());
+ processor.getPropertyDescriptors().clear();
+
+ for (final VersionedProcessGroup childGroup :
rootGroup.getProcessGroups()) {
+ if
(childGroup.getIdentifier().equals(processor.getGroupIdentifier())) {
+ childGroup.getConnections().removeIf(connection ->
connection.getSource().getId().equals(processor.getIdentifier()));
+ break;
+ }
+ }
+
+ return externalFlow;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/StandardConnectorNodeIT.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/StandardConnectorNodeIT.java
index 53ff90dab7c..7b28c75b011 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/StandardConnectorNodeIT.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/StandardConnectorNodeIT.java
@@ -28,6 +28,8 @@ import
org.apache.nifi.components.connector.processors.TerminateFlowFile;
import
org.apache.nifi.components.connector.secrets.ParameterProviderSecretsManager;
import org.apache.nifi.components.connector.secrets.SecretsManager;
import org.apache.nifi.components.connector.services.CounterService;
+import
org.apache.nifi.components.connector.services.impl.BlockingEnablingCounterService;
+import
org.apache.nifi.components.connector.services.impl.FailingEnablingCounterService;
import org.apache.nifi.components.state.StateManagerProvider;
import
org.apache.nifi.components.validation.StandardVerifiableComponentFactory;
import org.apache.nifi.components.validation.ValidationState;
@@ -42,6 +44,7 @@ import org.apache.nifi.controller.MockStateManagerProvider;
import org.apache.nifi.controller.NodeTypeProvider;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ReloadComponent;
+import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.flow.StandardFlowManager;
import org.apache.nifi.controller.flowanalysis.FlowAnalyzer;
import org.apache.nifi.controller.queue.DropFlowFileRequest;
@@ -60,8 +63,11 @@ import
org.apache.nifi.controller.scheduling.RepositoryContextFactory;
import org.apache.nifi.controller.scheduling.SchedulingAgent;
import org.apache.nifi.controller.scheduling.StandardLifecycleStateManager;
import org.apache.nifi.controller.scheduling.StandardProcessScheduler;
+import
org.apache.nifi.controller.scheduling.processors.FailOnScheduledProcessor;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
+import org.apache.nifi.controller.service.ControllerServiceState;
+import org.apache.nifi.controller.service.StandardControllerServiceProvider;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.mock.MockNodeTypeProvider;
@@ -93,11 +99,11 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.function.IntSupplier;
import java.util.function.Predicate;
import static java.util.Objects.requireNonNull;
@@ -121,11 +127,10 @@ public class StandardConnectorNodeIT {
private StandardFlowManager flowManager;
private FlowEngine componentLifecycleThreadPool;
private ConnectorRepository connectorRepository;
+ private ControllerServiceProvider controllerServiceProvider;
@BeforeEach
public void setup() {
- final ControllerServiceProvider controllerServiceProvider =
mock(ControllerServiceProvider.class);
-
when(controllerServiceProvider.disableControllerServicesAsync(anyCollection())).thenReturn(CompletableFuture.completedFuture(null));
connectorRepository = new StandardConnectorRepository();
final SecretsManager secretsManager = new
ParameterProviderSecretsManager();
@@ -139,7 +144,10 @@ public class StandardConnectorNodeIT {
final LifecycleStateManager lifecycleStateManager = new
StandardLifecycleStateManager();
final ReloadComponent reloadComponent = mock(ReloadComponent.class);
- final NiFiProperties nifiProperties =
NiFiProperties.createBasicNiFiProperties("src/test/resources/conf/nifi.properties");
+ final NiFiProperties nifiProperties =
NiFiProperties.createBasicNiFiProperties("src/test/resources/conf/nifi.properties",
Map.of(
+ NiFiProperties.ADMINISTRATIVE_YIELD_DURATION, "10 millis",
+ NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "30 secs"
+ ));
final FlowController flowController = mock(FlowController.class);
when(flowController.isInitialized()).thenReturn(true);
@@ -158,7 +166,6 @@ public class StandardConnectorNodeIT {
when(flowController.getRepositoryContextFactory()).thenReturn(repoContextFactory);
when(flowController.getGarbageCollectionLog()).thenReturn(mock(GarbageCollectionLog.class));
-
when(flowController.getControllerServiceProvider()).thenReturn(controllerServiceProvider);
when(flowController.getProvenanceRepository()).thenReturn(provRepo);
when(flowController.getBulletinRepository()).thenReturn(bulletinRepository);
when(flowController.getLifecycleStateManager()).thenReturn(lifecycleStateManager);
@@ -186,6 +193,8 @@ public class StandardConnectorNodeIT {
extensionManager.discoverExtensions(systemBundle, Set.of());
flowManager = new StandardFlowManager(nifiProperties, null,
flowController, flowFileEventRepository, parameterContextManager);
+ controllerServiceProvider = new
StandardControllerServiceProvider(processScheduler, bulletinRepository,
flowManager, extensionManager);
+
when(flowController.getControllerServiceProvider()).thenReturn(controllerServiceProvider);
flowManager.initialize(controllerServiceProvider,
mock(PythonBridge.class), mock(FlowAnalyzer.class),
mock(RuleViolationsManager.class));
final ProcessGroup rootGroup = flowManager.createProcessGroup("root");
rootGroup.setName("Root");
@@ -397,6 +406,234 @@ public class StandardConnectorNodeIT {
assertInstanceOf(CounterService.class,
serviceNodes.iterator().next().getControllerServiceImplementation());
}
+ @Test
+ public void testStopConnectorWhileManagedServiceBlocksInOnEnabled() throws
Exception {
+ final ConnectorNode connectorNode =
initializeControllerServiceEnablingConnector(ControllerServiceEnablingConnector.BLOCKING_ENABLING);
+ final ControllerServiceNode serviceNode =
getManagedControllerService(connectorNode);
+ final BlockingEnablingCounterService service =
(BlockingEnablingCounterService)
serviceNode.getControllerServiceImplementation();
+ service.setBlockDisable(true);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ try {
+ waitForServiceState(serviceNode, ControllerServiceState.ENABLING);
+ waitForEnableInvocation(service::enableInvocationCount, 1);
+
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ assertTrue(service.awaitEnableInterrupted(5, TimeUnit.SECONDS));
+ assertTrue(service.awaitDisableStarted(5, TimeUnit.SECONDS));
+ assertEquals(ControllerServiceState.DISABLING,
serviceNode.getState());
+ assertFalse(stopFuture.isDone());
+ assertEquals(1, service.disableInvocationCount());
+
+ service.releaseDisable();
+ stopFuture.get(5, TimeUnit.SECONDS);
+ assertEquals(ConnectorState.STOPPED,
connectorNode.getCurrentState());
+ assertEquals(ControllerServiceState.DISABLED,
serviceNode.getState());
+ } finally {
+ service.releaseFirstEnable();
+ service.releaseDisable();
+ }
+ }
+
+ @Test
+ public void testStopConnectorWhileManagedServiceKeepsFailingOnEnabled()
throws Exception {
+ final ConnectorNode connectorNode =
initializeControllerServiceEnablingConnector(ControllerServiceEnablingConnector.FAILING_ENABLING);
+ final ControllerServiceNode serviceNode =
getManagedControllerService(connectorNode);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ final FailingEnablingCounterService service =
(FailingEnablingCounterService)
serviceNode.getControllerServiceImplementation();
+ waitForEnableInvocation(service::enableInvocationCount, 2);
+
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ stopFuture.get(5, TimeUnit.SECONDS);
+
+ assertEquals(ConnectorState.STOPPED, connectorNode.getCurrentState());
+ assertEquals(ControllerServiceState.DISABLED, serviceNode.getState());
+
+ final int enableInvocationCount = service.enableInvocationCount();
+ Thread.sleep(100L);
+ assertEquals(enableInvocationCount, service.enableInvocationCount());
+ }
+
+ @Test
+ public void testRestartWaitsForPreviousManagedServiceEnableAndDisable()
throws Exception {
+ final ConnectorNode connectorNode =
initializeControllerServiceEnablingConnector(ControllerServiceEnablingConnector.BLOCKING_ENABLING);
+ final ControllerServiceNode serviceNode =
getManagedControllerService(connectorNode);
+ final BlockingEnablingCounterService service =
(BlockingEnablingCounterService)
serviceNode.getControllerServiceImplementation();
+ service.setIgnoreEnableInterrupt(true);
+ service.setBlockDisable(true);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ try {
+ waitForServiceState(serviceNode, ControllerServiceState.ENABLING);
+ waitForEnableInvocation(service::enableInvocationCount, 1);
+
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ final Future<Void> restartFuture =
connectorNode.start(componentLifecycleThreadPool);
+
+ assertTrue(service.awaitEnableInterrupted(5, TimeUnit.SECONDS));
+ waitForServiceState(serviceNode, ControllerServiceState.DISABLING);
+ assertFalse(stopFuture.isDone());
+ assertFalse(restartFuture.isDone());
+ assertEquals(1, service.enableInvocationCount());
+
+ service.releaseFirstEnable();
+ assertTrue(service.awaitDisableStarted(5, TimeUnit.SECONDS));
+ assertEquals(ControllerServiceState.DISABLING,
serviceNode.getState());
+ assertFalse(stopFuture.isDone());
+ assertFalse(restartFuture.isDone());
+ assertEquals(1, service.enableInvocationCount());
+
+ service.releaseDisable();
+ stopFuture.get(5, TimeUnit.SECONDS);
+ restartFuture.get(5, TimeUnit.SECONDS);
+ waitForEnableInvocation(service::enableInvocationCount, 2);
+ waitForServiceState(serviceNode, ControllerServiceState.ENABLED);
+
+ final List<String> lifecycleEvents = service.getLifecycleEvents();
+ assertTrue(lifecycleEvents.indexOf("enable-finished-1") <
lifecycleEvents.indexOf("disable-started-1"));
+ assertTrue(lifecycleEvents.indexOf("disable-finished-1") <
lifecycleEvents.indexOf("enable-started-2"));
+ } finally {
+ service.releaseFirstEnable();
+ service.releaseDisable();
+ }
+ }
+
+ @Test
+ public void
testStopConnectorWhileManagedProcessorKeepsFailingOnScheduled() throws
Exception {
+ final ConnectorNode connectorNode =
initializeProcessorStartFailureConnector();
+ final ProcessorNode processorNode = getManagedProcessor(connectorNode);
+ final FailOnScheduledProcessor processor = (FailOnScheduledProcessor)
processorNode.getProcessor();
+ processor.setDesiredFailureCount(Integer.MAX_VALUE);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ waitForEnableInvocation(processor::getOnScheduledInvocationCount, 2);
+
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ stopFuture.get(5, TimeUnit.SECONDS);
+
+ assertEquals(ConnectorState.STOPPED, connectorNode.getCurrentState());
+ assertEquals(ScheduledState.STOPPED,
processorNode.getPhysicalScheduledState());
+ assertEquals(ControllerServiceState.DISABLED,
getManagedControllerService(connectorNode).getState());
+
+ final int invocationCount = processor.getOnScheduledInvocationCount();
+ Thread.sleep(100L);
+ assertEquals(invocationCount,
processor.getOnScheduledInvocationCount());
+ }
+
+ @Test
+ public void testStopConnectorInterruptsManagedProcessorOnScheduled()
throws Exception {
+ final ConnectorNode connectorNode =
initializeProcessorStartFailureConnector();
+ final ProcessorNode processorNode = getManagedProcessor(connectorNode);
+ final FailOnScheduledProcessor processor = (FailOnScheduledProcessor)
processorNode.getProcessor();
+ processor.setDesiredFailureCount(0);
+ processor.setOnScheduledSleepDuration(20, TimeUnit.MINUTES, true, 1);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ waitForEnableInvocation(processor::getOnScheduledInvocationCount, 1);
+
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ stopFuture.get(5, TimeUnit.SECONDS);
+
+ assertEquals(ConnectorState.STOPPED, connectorNode.getCurrentState());
+ assertEquals(ScheduledState.STOPPED,
processorNode.getPhysicalScheduledState());
+ assertEquals(ControllerServiceState.DISABLED,
getManagedControllerService(connectorNode).getState());
+ assertFalse(processor.isSucceeded());
+ }
+
+ @Test
+ public void
testTerminateManagedProcessorWhoseOnScheduledIgnoresInterrupt() throws
Exception {
+ final ConnectorNode connectorNode =
initializeProcessorStartFailureConnector();
+ final ProcessorNode processorNode = getManagedProcessor(connectorNode);
+ final FailOnScheduledProcessor processor = (FailOnScheduledProcessor)
processorNode.getProcessor();
+ processor.setDesiredFailureCount(0);
+ processor.setOnScheduledSleepDuration(20, TimeUnit.MINUTES, false, 1);
+
+ connectorNode.start(componentLifecycleThreadPool);
+ waitForEnableInvocation(processor::getOnScheduledInvocationCount, 1);
+
+ try {
+ final Future<Void> stopFuture =
connectorNode.stop(componentLifecycleThreadPool);
+ waitForProcessorState(processorNode, ScheduledState.STOPPING);
+ Thread.sleep(100L);
+
+ assertFalse(stopFuture.isDone());
+ assertEquals(ConnectorState.STOPPING,
connectorNode.getCurrentState());
+ assertEquals(ControllerServiceState.ENABLED,
getManagedControllerService(connectorNode).getState());
+
+ processor.setAllowSleepInterrupt(true);
+ processorNode.getProcessGroup().terminateProcessor(processorNode);
+ stopFuture.get(5, TimeUnit.SECONDS);
+
+ assertEquals(ConnectorState.STOPPED,
connectorNode.getCurrentState());
+ assertEquals(ScheduledState.STOPPED,
processorNode.getPhysicalScheduledState());
+ assertEquals(ControllerServiceState.DISABLED,
getManagedControllerService(connectorNode).getState());
+ } finally {
+ processor.setAllowSleepInterrupt(true);
+ }
+ }
+
+ private ConnectorNode initializeControllerServiceEnablingConnector(final
String enablingBehavior) throws FlowUpdateException {
+ final ConnectorNode connectorNode =
flowManager.createConnector(ControllerServiceEnablingConnector.class.getName(),
+ "controller-service-enabling-connector",
SystemBundle.SYSTEM_BUNDLE_COORDINATE, true, true);
+
+ final StepConfiguration stepConfiguration = new
StepConfiguration(Map.of(
+ ControllerServiceEnablingConnector.ENABLING_BEHAVIOR.getName(),
new StringLiteralValue(enablingBehavior)));
+
+ final NamedStepConfiguration namedStepConfiguration = new
NamedStepConfiguration(ControllerServiceEnablingConnector.CONFIGURATION_STEP_NAME,
stepConfiguration);
+ configure(connectorNode, new
ConnectorConfiguration(Set.of(namedStepConfiguration)));
+ return connectorNode;
+ }
+
+ private ControllerServiceNode getManagedControllerService(final
ConnectorNode connectorNode) {
+ final ProcessGroup managedGroup =
connectorNode.getActiveFlowContext().getManagedProcessGroup();
+ final Set<ControllerServiceNode> services =
managedGroup.getControllerServices(true);
+ assertEquals(1, services.size());
+ return services.iterator().next();
+ }
+
+ private ConnectorNode initializeProcessorStartFailureConnector() {
+ final ConnectorNode connectorNode =
flowManager.createConnector(ProcessorStartFailureConnector.class.getName(),
+ "processor-start-failure-connector",
SystemBundle.SYSTEM_BUNDLE_COORDINATE, true, true);
+ assertNotNull(connectorNode);
+ return connectorNode;
+ }
+
+ private ProcessorNode getManagedProcessor(final ConnectorNode
connectorNode) {
+ final ProcessGroup managedGroup =
connectorNode.getActiveFlowContext().getManagedProcessGroup();
+ return managedGroup.findAllProcessors().stream()
+ .filter(processorNode -> processorNode.getProcessor() instanceof
FailOnScheduledProcessor)
+ .findFirst()
+ .orElseThrow();
+ }
+
+ private void waitForServiceState(final ControllerServiceNode serviceNode,
final ControllerServiceState desiredState) throws InterruptedException {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (serviceNode.getState() != desiredState && System.nanoTime() <
deadline) {
+ Thread.sleep(10L);
+ }
+
+ assertEquals(desiredState, serviceNode.getState());
+ }
+
+ private void waitForEnableInvocation(final IntSupplier
enableInvocationCount, final int expectedCount) throws InterruptedException {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(15);
+ while (enableInvocationCount.getAsInt() < expectedCount &&
System.nanoTime() < deadline) {
+ Thread.sleep(10L);
+ }
+
+ assertTrue(enableInvocationCount.getAsInt() >= expectedCount);
+ }
+
+ private void waitForProcessorState(final ProcessorNode processorNode,
final ScheduledState desiredState) throws InterruptedException {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ while (processorNode.getPhysicalScheduledState() != desiredState &&
System.nanoTime() < deadline) {
+ Thread.sleep(10L);
+ }
+
+ assertEquals(desiredState, processorNode.getPhysicalScheduledState());
+ }
+
@Test
public void testUpdateProcessorPropertyDataQueued() throws
FlowUpdateException {
final ConnectorNode connectorNode = initializeDynamicFlowConnector();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/processors/CreateDummyFlowFile.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/processors/CreateDummyFlowFile.java
index 98ead292e1b..3e7abd37285 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/processors/CreateDummyFlowFile.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/processors/CreateDummyFlowFile.java
@@ -57,7 +57,7 @@ public class CreateDummyFlowFile extends AbstractProcessor {
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return List.of(TEXT);
+ return List.of(TEXT, COUNT_SERVICE);
}
@Override
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/BlockingEnablingCounterService.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/BlockingEnablingCounterService.java
new file mode 100644
index 00000000000..3fd80f01047
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/BlockingEnablingCounterService.java
@@ -0,0 +1,122 @@
+/*
+ * 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.services.impl;
+
+import org.apache.nifi.annotation.lifecycle.OnDisabled;
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.connector.services.CounterService;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class BlockingEnablingCounterService extends AbstractControllerService
implements CounterService {
+
+ private final AtomicInteger enableCounter = new AtomicInteger();
+ private final AtomicInteger disableCounter = new AtomicInteger();
+ private final CountDownLatch firstEnableRelease = new CountDownLatch(1);
+ private final CountDownLatch enableInterrupted = new CountDownLatch(1);
+ private final CountDownLatch disableStarted = new CountDownLatch(1);
+ private final CountDownLatch disableRelease = new CountDownLatch(1);
+ private final List<String> lifecycleEvents = new CopyOnWriteArrayList<>();
+ private volatile boolean ignoreEnableInterrupt;
+ private volatile boolean blockDisable;
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) throws
InterruptedException {
+ final int invocation = enableCounter.incrementAndGet();
+ lifecycleEvents.add("enable-started-" + invocation);
+ if (invocation == 1) {
+ while (firstEnableRelease.getCount() > 0) {
+ try {
+ firstEnableRelease.await();
+ } catch (final InterruptedException e) {
+ enableInterrupted.countDown();
+ lifecycleEvents.add("enable-interrupted-" + invocation);
+ if (!ignoreEnableInterrupt) {
+ throw e;
+ }
+ }
+ }
+ }
+
+ lifecycleEvents.add("enable-finished-" + invocation);
+ }
+
+ @OnDisabled
+ public void onDisabled(final ConfigurationContext context) throws
InterruptedException {
+ final int invocation = disableCounter.incrementAndGet();
+ lifecycleEvents.add("disable-started-" + invocation);
+ disableStarted.countDown();
+ if (blockDisable) {
+ disableRelease.await();
+ }
+
+ lifecycleEvents.add("disable-finished-" + invocation);
+ }
+
+ public int enableInvocationCount() {
+ return enableCounter.get();
+ }
+
+ public int disableInvocationCount() {
+ return disableCounter.get();
+ }
+
+ public void setIgnoreEnableInterrupt(final boolean ignoreEnableInterrupt) {
+ this.ignoreEnableInterrupt = ignoreEnableInterrupt;
+ }
+
+ public void setBlockDisable(final boolean blockDisable) {
+ this.blockDisable = blockDisable;
+ }
+
+ public boolean awaitEnableInterrupted(final long timeout, final TimeUnit
timeUnit) throws InterruptedException {
+ return enableInterrupted.await(timeout, timeUnit);
+ }
+
+ public boolean awaitDisableStarted(final long timeout, final TimeUnit
timeUnit) throws InterruptedException {
+ return disableStarted.await(timeout, timeUnit);
+ }
+
+ public List<String> getLifecycleEvents() {
+ return List.copyOf(lifecycleEvents);
+ }
+
+ public void releaseFirstEnable() {
+ firstEnableRelease.countDown();
+ }
+
+ public void releaseDisable() {
+ disableRelease.countDown();
+ }
+
+ @Override
+ public long increment() {
+ return 0;
+ }
+
+ @Override
+ public long getCount() {
+ return 0;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/FailingEnablingCounterService.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/FailingEnablingCounterService.java
new file mode 100644
index 00000000000..b61c8b1b114
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/connector/services/impl/FailingEnablingCounterService.java
@@ -0,0 +1,50 @@
+/*
+ * 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.services.impl;
+
+import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.connector.services.CounterService;
+import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class FailingEnablingCounterService extends AbstractControllerService
implements CounterService {
+
+ private final AtomicInteger enableCounter = new AtomicInteger();
+
+ @OnEnabled
+ public void onEnabled(final ConfigurationContext context) {
+ enableCounter.incrementAndGet();
+ throw new IllegalStateException("Configured to always fail
enablement");
+ }
+
+ public int enableInvocationCount() {
+ return enableCounter.get();
+ }
+
+ @Override
+ public long increment() {
+ return 0;
+ }
+
+ @Override
+ public long getCount() {
+ return 0;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
index 0b90dc5985d..e45c13857e5 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java
@@ -27,6 +27,7 @@ import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.components.validation.ValidationTrigger;
import org.apache.nifi.components.validation.VerifiableComponentFactory;
+import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ExtensionBuilder;
@@ -95,13 +96,16 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+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.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -110,7 +114,9 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertSame;
+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;
@@ -503,6 +509,36 @@ public class TestStandardProcessScheduler {
assertEquals(0, ts.disableInvocationCount());
}
+ @Test
+ @Timeout(10)
+ public void testRepeatedDisableWhileEnablingCompletesFutures() throws
Exception {
+ final StandardProcessScheduler scheduler = createScheduler();
+ final ControllerServiceNode serviceNode =
flowManager.createControllerService(LongEnablingService.class.getName(),
+ "1", systemBundle.getBundleDetails().getCoordinate(), null,
false, true, null);
+ final LongEnablingService service = (LongEnablingService)
serviceNode.getControllerServiceImplementation();
+ service.setLimit(TimeUnit.SECONDS.toMillis(1));
+
+ serviceNode.performValidation();
+ final CompletableFuture<Void> enableFuture =
scheduler.enableControllerService(serviceNode);
+
+ final long enableInvocationDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (service.enableInvocationCount() == 0 && System.nanoTime() <
enableInvocationDeadline) {
+ Thread.sleep(1L);
+ }
+
+ assertEquals(1, service.enableInvocationCount());
+ assertEquals(ControllerServiceState.ENABLING, serviceNode.getState());
+
+ final CompletableFuture<Void> firstDisableFuture =
scheduler.disableControllerService(serviceNode);
+ final CompletableFuture<Void> secondDisableFuture =
scheduler.disableControllerService(serviceNode);
+
+ final ExecutionException enableFailure =
assertThrows(ExecutionException.class, () -> enableFuture.get(5,
TimeUnit.SECONDS));
+ assertInstanceOf(CancellationException.class,
enableFailure.getCause());
+ firstDisableFuture.get(5, TimeUnit.SECONDS);
+ secondDisableFuture.get(5, TimeUnit.SECONDS);
+ assertEquals(ControllerServiceState.DISABLED, serviceNode.getState());
+ }
+
@Test
@Timeout(10)
public void testEnableControllerServiceWithConfigurationContext() throws
Exception {
@@ -548,6 +584,61 @@ public class TestStandardProcessScheduler {
assertEquals(ControllerServiceState.ENABLED, serviceNode.getState());
}
+ @Test
+ public void testProcessorStopWaitsForSchedulingAgentUnschedule() throws
Exception {
+ final CountDownLatch schedulingStarted = new CountDownLatch(1);
+ final Semaphore schedulingRelease = new Semaphore(0);
+ final CountDownLatch unschedulingStarted = new CountDownLatch(1);
+ final Semaphore unschedulingRelease = new Semaphore(0);
+ final SchedulingAgent schedulingAgent = mock(SchedulingAgent.class);
+
+ doAnswer(invocation -> scheduleAgent(true, invocation.getArgument(1),
schedulingStarted, schedulingRelease)).when(schedulingAgent)
+ .schedule(any(Connectable.class), any(LifecycleState.class));
+ doAnswer(invocation -> scheduleAgent(false, invocation.getArgument(1),
unschedulingStarted, unschedulingRelease)).when(schedulingAgent)
+ .unschedule(any(Connectable.class), any(LifecycleState.class));
+ scheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN,
schedulingAgent);
+
+ final String identifier = UUID.randomUUID().toString();
+ final Processor processor = new NoOpProcessor();
+ processor.initialize(new
StandardProcessorInitializationContext(identifier, null, null, null,
KerberosConfig.NOT_CONFIGURED));
+ final LoggableComponent<Processor> loggableComponent = new
LoggableComponent<>(processor, systemBundle.getBundleDetails().getCoordinate(),
null);
+ final ProcessorNode processorNode = new
StandardProcessorNode(loggableComponent, identifier, new
StandardValidationContextFactory(serviceProvider), scheduler,
+ serviceProvider, mock(ReloadComponent.class),
mock(VerifiableComponentFactory.class), extensionManager, new
SynchronousValidationTrigger());
+ rootGroup.addProcessor(processorNode);
+ processorNode.performValidation();
+
+ // Hold the startup task inside SchedulingAgent.schedule() after the
Processor transitions to RUNNING.
+ scheduler.startProcessor(processorNode, true);
+ assertTrue(schedulingStarted.await(5, TimeUnit.SECONDS));
+
+ // Start the stop sequence while the startup task is still returning
from the scheduling callback.
+ final CompletableFuture<Void> stopFuture =
scheduler.stopProcessor(processorNode,
ProcessorStopLifecycleMethods.TRIGGER_ALL);
+
+ try {
+ // Allow startup to return, then hold the stop task inside
SchedulingAgent.unschedule().
+ schedulingRelease.release();
+ assertTrue(unschedulingStarted.await(5, TimeUnit.SECONDS));
+
+ // The Processor cannot report that it is stopped while
unscheduling is still running.
+ assertFalse(stopFuture.isDone());
+ } finally {
+ schedulingRelease.release();
+ unschedulingRelease.release();
+ }
+
+ // Releasing the unschedule callback allows the normal stop sequence
to finish.
+ stopFuture.get(5, TimeUnit.SECONDS);
+ assertEquals(ScheduledState.STOPPED,
processorNode.getPhysicalScheduledState());
+ }
+
+ private Void scheduleAgent(final boolean scheduled, final LifecycleState
lifecycleState,
+ final CountDownLatch callbackStarted, final Semaphore
callbackRelease) {
+ lifecycleState.setScheduled(scheduled);
+ callbackStarted.countDown();
+ callbackRelease.acquireUninterruptibly();
+ return null;
+ }
+
// Test that if processor throws Exception in @OnScheduled, it keeps
getting scheduled
@Test
@Timeout(10)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/processors/FailOnScheduledProcessor.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/processors/FailOnScheduledProcessor.java
index 601b600ccc3..6590499c5f9 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/processors/FailOnScheduledProcessor.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/processors/FailOnScheduledProcessor.java
@@ -18,22 +18,33 @@
package org.apache.nifi.controller.scheduling.processors;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.connector.services.CounterService;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
public class FailOnScheduledProcessor extends AbstractProcessor {
- private volatile int invocationCount = 0;
+ public static final PropertyDescriptor MANAGED_SERVICE = new
PropertyDescriptor.Builder()
+ .name("Managed Service")
+ .description("Managed Controller Service used to verify lifecycle
ordering")
+ .identifiesControllerService(CounterService.class)
+ .required(false)
+ .build();
+
+ private final AtomicInteger invocationCount = new AtomicInteger();
+ private final AtomicBoolean succeeded = new AtomicBoolean();
private volatile int desiredFailureCount = 1;
private volatile long onScheduledSleepMillis = 0L;
private volatile int onScheduledSleepIterations = 0;
private volatile boolean allowSleepInterrupt = true;
- private final AtomicBoolean succeeded = new AtomicBoolean();
public void setDesiredFailureCount(final int desiredFailureCount) {
this.desiredFailureCount = desiredFailureCount;
@@ -51,26 +62,24 @@ public class FailOnScheduledProcessor extends
AbstractProcessor {
@OnScheduled
public void onScheduled() throws InterruptedException {
- invocationCount++;
+ final int invocation = invocationCount.incrementAndGet();
- if (invocationCount <= onScheduledSleepIterations &&
onScheduledSleepMillis > 0L) {
+ if (invocation <= onScheduledSleepIterations && onScheduledSleepMillis
> 0L) {
final long sleepFinish = System.currentTimeMillis() +
onScheduledSleepMillis;
while (System.currentTimeMillis() < sleepFinish) {
try {
Thread.sleep(Math.max(0, sleepFinish -
System.currentTimeMillis()));
- } catch (final InterruptedException ie) {
+ } catch (final InterruptedException e) {
if (allowSleepInterrupt) {
Thread.currentThread().interrupt();
- throw ie;
- } else {
- continue;
+ throw e;
}
}
}
}
- if (invocationCount < desiredFailureCount) {
+ if (invocation < desiredFailureCount) {
throw new ProcessException("Intentional failure for unit test");
} else {
succeeded.set(true);
@@ -78,7 +87,7 @@ public class FailOnScheduledProcessor extends
AbstractProcessor {
}
public int getOnScheduledInvocationCount() {
- return invocationCount;
+ return invocationCount.get();
}
public boolean isSucceeded() {
@@ -86,6 +95,11 @@ public class FailOnScheduledProcessor extends
AbstractProcessor {
}
@Override
- public void onTrigger(ProcessContext context, ProcessSession session)
throws ProcessException {
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return List.of(MANAGED_SERVICE);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession
session) throws ProcessException {
}
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
index 41664bae145..371e8edd05d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
@@ -71,7 +71,10 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -94,6 +97,7 @@ public class TestStandardControllerServiceProvider {
private static ExtensionDiscoveringManager extensionManager;
private static Bundle systemBundle;
private FlowManager flowManager;
+ private ReloadComponent reloadComponent;
@BeforeAll
public static void setNiFiProps() {
@@ -110,6 +114,7 @@ public class TestStandardControllerServiceProvider {
@BeforeEach
public void setup() {
flowManager = mock(FlowManager.class);
+ reloadComponent = mock(ReloadComponent.class);
final ConcurrentMap<String, ProcessorNode> processorMap = new
ConcurrentHashMap<>();
doAnswer((Answer<ProcessorNode>) invocation -> {
@@ -144,7 +149,7 @@ public class TestStandardControllerServiceProvider {
.processScheduler(mock(ProcessScheduler.class))
.nodeTypeProvider(mock(NodeTypeProvider.class))
.validationTrigger(mock(ValidationTrigger.class))
- .reloadComponent(mock(ReloadComponent.class))
+ .reloadComponent(reloadComponent)
.verifiableComponentFactory(mock(VerifiableComponentFactory.class))
.stateManagerProvider(mock(StateManagerProvider.class))
.extensionManager(extensionManager)
@@ -172,6 +177,53 @@ public class TestStandardControllerServiceProvider {
provider.disableControllerService(serviceNode);
}
+ @Test
+ public void testEnableAfterConcurrentDisable() throws Exception {
+ final StandardProcessScheduler scheduler = createScheduler();
+ final StandardControllerServiceProvider provider = new
StandardControllerServiceProvider(scheduler, null, flowManager,
extensionManager);
+ final ControllerServiceNode serviceNode =
createControllerService(ServiceB.class.getName(), "B",
systemBundle.getBundleDetails().getCoordinate(), provider);
+ final ScheduledExecutorService lifecycleExecutor =
Executors.newScheduledThreadPool(2);
+ final AtomicReference<CompletableFuture<Void>> disableFutureReference
= new AtomicReference<>();
+ final AtomicReference<Thread> enableThreadReference = new
AtomicReference<>();
+
+ // Reloading holds the lifecycle monitor. Start enabling on another
thread so that disabling can be requested before the monitor is released.
+ doAnswer(invocation -> {
+ final Thread enableThread = new Thread(() ->
serviceNode.enable(lifecycleExecutor, 10, true), "Controller Service Enable");
+ enableThreadReference.set(enableThread);
+ enableThread.start();
+
+ final long waitDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(5);
+ while (enableThread.getState() != Thread.State.BLOCKED &&
System.nanoTime() < waitDeadline) {
+ Thread.sleep(10);
+ }
+ assertEquals(Thread.State.BLOCKED, enableThread.getState());
+
+ disableFutureReference.set(serviceNode.disable(lifecycleExecutor));
+ return null;
+ }).when(reloadComponent).reload(any(ControllerServiceNode.class),
anyString(), any(BundleCoordinate.class), any());
+
+ try {
+ serviceNode.performValidation();
+ assertEquals(ValidationStatus.VALID,
serviceNode.getValidationStatus(5, TimeUnit.SECONDS));
+
+ serviceNode.reload(Collections.emptySet());
+
+ final Thread enableThread = enableThreadReference.get();
+ enableThread.join(TimeUnit.SECONDS.toMillis(5));
+ assertFalse(enableThread.isAlive());
+ disableFutureReference.get().get(5, TimeUnit.SECONDS);
+
+ // A later enable request must remain effective after the
concurrent enable and disable operations finish.
+ provider.enableControllerService(serviceNode);
+
+ assertTrue(serviceNode.awaitEnabled(5, TimeUnit.SECONDS));
+ assertEquals(ControllerServiceState.ENABLED,
serviceNode.getState());
+ } finally {
+ lifecycleExecutor.shutdownNow();
+ scheduler.shutdown();
+ }
+ }
+
@Test
@Timeout(10)
public void testEnableDisableWithReference() {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.components.connector.Connector
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.components.connector.Connector
index 93fb310b0ac..8bd1dfec594 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.components.connector.Connector
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.components.connector.Connector
@@ -16,3 +16,5 @@
org.apache.nifi.controller.flow.NopConnector
org.apache.nifi.components.connector.DynamicFlowConnector
org.apache.nifi.components.connector.DynamicAllowableValuesConnector
+org.apache.nifi.components.connector.ControllerServiceEnablingConnector
+org.apache.nifi.components.connector.ProcessorStartFailureConnector
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.controller.ControllerService
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.controller.ControllerService
index d9e0a9a8385..692e0763f98 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.controller.ControllerService
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.controller.ControllerService
@@ -19,4 +19,6 @@ org.apache.nifi.controller.service.mock.ServiceB
org.apache.nifi.controller.service.mock.ServiceC
org.apache.nifi.controller.service.mock.ServiceD
-org.apache.nifi.components.connector.services.impl.StandardCounterService
\ No newline at end of file
+org.apache.nifi.components.connector.services.impl.StandardCounterService
+org.apache.nifi.components.connector.services.impl.BlockingEnablingCounterService
+org.apache.nifi.components.connector.services.impl.FailingEnablingCounterService
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.processor.Processor
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.processor.Processor
index ff061c32002..3d0e8a7a63f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.processor.Processor
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -23,4 +23,5 @@
org.apache.nifi.components.connector.processors.TerminateFlowFile
org.apache.nifi.components.connector.processors.LogFlowFileContents
org.apache.nifi.components.connector.processors.ExposeFileValues
org.apache.nifi.components.connector.processors.Sleep
-org.apache.nifi.components.connector.processors.OnPropertyModifiedTracker
\ No newline at end of file
+org.apache.nifi.components.connector.processors.OnPropertyModifiedTracker
+org.apache.nifi.controller.scheduling.processors.FailOnScheduledProcessor
\ No newline at end of file
