Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-1464, Refactored Processor's life-cycle operation sequence 

* Simplified and cleaned StandardProcessScheduler.start/stopProcessor methods
* Added stop/start operations to ProcessorNode.
* Removed unnecessary synchronization blocks related to ScheduledState in favor of enforcing order and idempotency via CAS operations. Those synchronization blocks were causing intermittent deadlocks whenever @OnScheduled blocks indefinitely.
* Added support for stopping the service when @OnScheduled operation hangs.
* Fixed the order of life-cycle operation invocation ensuring that each operation can *only* be invoked at the appropriate time
* Removed unnecessary locks from StandardProcessNode since Atomic variables are used.
* Removed calls to @OnStopped from ContinuallyRunningProcessTask while ensuring that procesor's full shut down in implementation of StandardProcessorNode.stop() method.
* Removed dead code
* Added comprehensive tests suite that covers 95% of Processor's life-cycle operations within the scope of FlowController, StandardProcesssScheduler and StandardProcessNode
* Improved and added javadocs on covered operations with detailed explanations.
This commit is contained in:
Oleg Zhurakousky 2016-02-04 16:23:45 -05:00 committed by Mark Payne
parent f6705f234c
commit 0c5b1c27f2
17 changed files with 1665 additions and 747 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java
View File

@ -33,5 +33,9 @@ public enum ScheduledState {
/**
* Entity is currently scheduled to run
*/
RUNNING;
RUNNING,
STARTING,
STOPPING;
}

2
nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
View File

@ -71,6 +71,7 @@ public class NiFiProperties extends Properties {
public static final String ADMINISTRATIVE_YIELD_DURATION = "nifi.administrative.yield.duration";
public static final String PERSISTENT_STATE_DIRECTORY = "nifi.persistent.state.directory";
public static final String BORED_YIELD_DURATION = "nifi.bored.yield.duration";
public static final String PROCESSOR_START_TIMEOUT = "nifi.processor.start.timeout";
// content repository properties
public static final String REPOSITORY_CONTENT_PREFIX = "nifi.content.repository.directory.";
@ -539,6 +540,7 @@ public class NiFiProperties extends Properties {
return shouldSupport;
}
@SuppressWarnings("unchecked")
public Set<String> getAnonymousAuthorities() {
final Set<String> authorities;

1
nifi-docs/src/main/asciidoc/administration-guide.adoc
View File

@ -1153,6 +1153,7 @@ nifi.nar.library.directory.lib2=/nars/lib2 +
Providing three total locations, including _nifi.nar.library.directory_.
|nifi.nar.working.directory|The location of the nar working directory. The default value is ./work/nar and probably should be left as is.
|nifi.documentation.working.directory|The documentation working directory. The default value is ./work/docs/components and probably should be left as is.
|nifi.processor.start.timeout|Time (milliseconds) to wait for a Processors to start before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
|====

7
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
View File

@ -44,7 +44,6 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone
private final ConfigurableComponent component;
private final ValidationContextFactory validationContextFactory;
private final ControllerServiceProvider serviceProvider;
private final AtomicReference<String> name;
private final AtomicReference<String> annotationData = new AtomicReference<>();
@ -298,4 +297,10 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone
public abstract void verifyModifiable() throws IllegalStateException;
/**
*
*/
ControllerServiceProvider getControllerServiceProvider() {
return this.serviceProvider;
}
}

50
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java
View File

@ -18,6 +18,8 @@ package org.apache.nifi.controller;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
@ -25,6 +27,7 @@ import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.scheduling.SchedulingStrategy;
@ -54,8 +57,6 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
@Override
public abstract boolean isValid();
public abstract void setScheduledState(ScheduledState scheduledState);
public abstract void setBulletinLevel(LogLevel bulletinLevel);
public abstract LogLevel getBulletinLevel();
@ -99,4 +100,49 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
*/
public abstract void verifyCanStart(Set<ControllerServiceNode> ignoredReferences);
/**
* Will start the {@link Processor} represented by this
* {@link ProcessorNode}. Starting processor typically means invoking it's
* operation that is annotated with @OnScheduled and then executing a
* callback provided by the {@link ProcessScheduler} to which typically
* initiates
* {@link Processor#onTrigger(ProcessContext, org.apache.nifi.processor.ProcessSessionFactory)}
* cycle.
*
* @param scheduler
* implementation of {@link ScheduledExecutorService} used to
* initiate processor <i>start</i> task
* @param administrativeYieldMillis
* the amount of milliseconds to wait for administrative yield
* @param processContext
* the instance of {@link ProcessContext} and
* {@link ControllerServiceLookup}
* @param schedulingAgentCallback
* the callback provided by the {@link ProcessScheduler} to
* execute upon successful start of the Processor
*/
public abstract <T extends ProcessContext & ControllerServiceLookup> void start(ScheduledExecutorService scheduler,
long administrativeYieldMillis, T processContext, Runnable schedulingAgentCallback);
/**
* Will stop the {@link Processor} represented by this {@link ProcessorNode}.
* Stopping processor typically means invoking it's operation that is
* annotated with @OnUnschedule and then @OnStopped.
*
* @param scheduler
* implementation of {@link ScheduledExecutorService} used to
* initiate processor <i>stop</i> task
* @param processContext
* the instance of {@link ProcessContext} and
* {@link ControllerServiceLookup}
* @param activeThreadMonitorCallback
* the callback provided by the {@link ProcessScheduler} to
* report the count of processor's active threads. Typically it
* is used to ensure that operations annotated with @OnUnschedule
* and then @OnStopped are not invoked until such count reaches
* 0, essentially allowing tasks to finish before bringing
* processor to a halt.
*/
public abstract <T extends ProcessContext & ControllerServiceLookup> void stop(ScheduledExecutorService scheduler,
T processContext, Callable<Boolean> activeThreadMonitorCallback);
}

1155
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java
View File

File diff suppressed because it is too large Load Diff

103
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java Normal file
View File

@ -0,0 +1,103 @@
/*
* 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.controller.scheduling;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.controller.ReportingTaskNode;
/**
* Base implementation of the {@link SchedulingAgent} which encapsulates the
* updates to the {@link ScheduleState} based on invoked operation and then
* delegates to the corresponding 'do' methods. For example; By invoking
* {@link #schedule(Connectable, ScheduleState)} the the
* {@link ScheduleState#setScheduled(boolean)} with value 'true' will be
* invoked.
*
* @see EventDrivenSchedulingAgent
* @see TimerDrivenSchedulingAgent
* @see QuartzSchedulingAgent
*/
abstract class AbstractSchedulingAgent implements SchedulingAgent {
@Override
public void schedule(Connectable connectable, ScheduleState scheduleState) {
scheduleState.setScheduled(true);
this.doSchedule(connectable, scheduleState);
}
@Override
public void unschedule(Connectable connectable, ScheduleState scheduleState) {
scheduleState.setScheduled(false);
this.doUnschedule(connectable, scheduleState);
}
@Override
public void schedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
scheduleState.setScheduled(true);
this.doSchedule(taskNode, scheduleState);
}
@Override
public void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
scheduleState.setScheduled(false);
this.doUnschedule(taskNode, scheduleState);
}
/**
* Schedules the provided {@link Connectable}. Its {@link ScheduleState}
* will be set to <i>true</i>
*
* @param connectable
* the instance of {@link Connectable}
* @param scheduleState
* the instance of {@link ScheduleState}
*/
protected abstract void doSchedule(Connectable connectable, ScheduleState scheduleState);
/**
* Unschedules the provided {@link Connectable}. Its {@link ScheduleState}
* will be set to <i>false</i>
*
* @param connectable
* the instance of {@link Connectable}
* @param scheduleState
* the instance of {@link ScheduleState}
*/
protected abstract void doUnschedule(Connectable connectable, ScheduleState scheduleState);
/**
* Schedules the provided {@link ReportingTaskNode}. Its
* {@link ScheduleState} will be set to <i>true</i>
*
* @param connectable
* the instance of {@link ReportingTaskNode}
* @param scheduleState
* the instance of {@link ScheduleState}
*/
protected abstract void doSchedule(ReportingTaskNode connectable, ScheduleState scheduleState);
/**
* Unschedules the provided {@link ReportingTaskNode}. Its
* {@link ScheduleState} will be set to <i>false</i>
*
* @param connectable
* the instance of {@link ReportingTaskNode}
* @param scheduleState
* the instance of {@link ScheduleState}
*/
protected abstract void doUnschedule(ReportingTaskNode connectable, ScheduleState scheduleState);
}

10
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
View File

@ -51,7 +51,7 @@ import org.apache.nifi.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class EventDrivenSchedulingAgent implements SchedulingAgent {
public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
private static final Logger logger = LoggerFactory.getLogger(EventDrivenSchedulingAgent.class);
@ -94,24 +94,24 @@ public class EventDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
public void schedule(final ReportingTaskNode taskNode, ScheduleState scheduleState) {
public void doSchedule(final ReportingTaskNode taskNode, ScheduleState scheduleState) {
throw new UnsupportedOperationException("ReportingTasks cannot be scheduled in Event-Driven Mode");
}
@Override
public void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
public void doUnschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
throw new UnsupportedOperationException("ReportingTasks cannot be scheduled in Event-Driven Mode");
}
@Override
public void schedule(final Connectable connectable, final ScheduleState scheduleState) {
public void doSchedule(final Connectable connectable, final ScheduleState scheduleState) {
workerQueue.resumeWork(connectable);
logger.info("Scheduled {} to run in Event-Driven mode", connectable);
scheduleStates.put(connectable, scheduleState);
}
@Override
public void unschedule(final Connectable connectable, final ScheduleState scheduleState) {
public void doUnschedule(final Connectable connectable, final ScheduleState scheduleState) {
workerQueue.suspendWork(connectable);
logger.info("Stopped scheduling {} to run", connectable);
}

10
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
View File

@ -43,7 +43,7 @@ import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class QuartzSchedulingAgent implements SchedulingAgent {
public class QuartzSchedulingAgent extends AbstractSchedulingAgent {
private final Logger logger = LoggerFactory.getLogger(QuartzSchedulingAgent.class);
@ -71,7 +71,7 @@ public class QuartzSchedulingAgent implements SchedulingAgent {
}
@Override
public void schedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
public void doSchedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
final List<AtomicBoolean> existingTriggers = canceledTriggers.get(taskNode);
if (existingTriggers != null) {
throw new IllegalStateException("Cannot schedule " + taskNode.getReportingTask() + " because it is already scheduled to run");
@ -121,7 +121,7 @@ public class QuartzSchedulingAgent implements SchedulingAgent {
}
@Override
public synchronized void schedule(final Connectable connectable, final ScheduleState scheduleState) {
public synchronized void doSchedule(final Connectable connectable, final ScheduleState scheduleState) {
final List<AtomicBoolean> existingTriggers = canceledTriggers.get(connectable);
if (existingTriggers != null) {
throw new IllegalStateException("Cannot schedule " + connectable + " because it is already scheduled to run");
@ -189,12 +189,12 @@ public class QuartzSchedulingAgent implements SchedulingAgent {
}
@Override
public synchronized void unschedule(final Connectable connectable, final ScheduleState scheduleState) {
public synchronized void doUnschedule(final Connectable connectable, final ScheduleState scheduleState) {
unschedule((Object) connectable, scheduleState);
}
@Override
public synchronized void unschedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
public synchronized void doUnschedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
unschedule((Object) taskNode, scheduleState);
}

8
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java
View File

@ -50,7 +50,7 @@ public class ScheduleState {
return scheduled.get();
}
public void setScheduled(final boolean scheduled) {
void setScheduled(final boolean scheduled) {
this.scheduled.set(scheduled);
mustCallOnStoppedMethods.set(true);
@ -63,6 +63,12 @@ public class ScheduleState {
return lastStopTime;
}
@Override
public String toString() {
return new StringBuilder().append("activeThreads:").append(activeThreadCount.get()).append("; ")
.append("scheduled:").append(scheduled.get()).append("; ").toString();
}
/**
* Maintains an AtomicBoolean so that the first thread to call this method after a Processor is no longer
* scheduled to run will receive a <code>true</code> and MUST call the methods annotated with

220
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
View File

@ -19,9 +19,8 @@ package org.apache.nifi.controller.scheduling;
import static java.util.Objects.requireNonNull;
import java.lang.reflect.InvocationTargetException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
@ -31,7 +30,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.connectable.Connectable;
@ -39,24 +37,22 @@ import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.controller.AbstractPort;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.Heartbeater;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ReportingTaskNode;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.StandardProcessorNode;
import org.apache.nifi.controller.annotation.OnConfigured;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.logging.ProcessorLog;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.SchedulingContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.StandardSchedulingContext;
import org.apache.nifi.reporting.ReportingTask;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.FormatUtils;
@ -288,173 +284,76 @@ public final class StandardProcessScheduler implements ProcessScheduler {
}
/**
* Starts scheduling the given processor to run after invoking all methods on the underlying {@link org.apache.nifi.processor.Processor
* FlowFileProcessor} that are annotated with the {@link OnScheduled} annotation.
* Starts the given {@link Processor} by invoking its
* {@link ProcessorNode#start(ScheduledExecutorService, long, org.apache.nifi.processor.ProcessContext, Runnable)}
* .
* @see StandardProcessorNode#start(ScheduledExecutorService, long,
* org.apache.nifi.processor.ProcessContext, Runnable).
*/
@Override
public synchronized void startProcessor(final ProcessorNode procNode) {
if (procNode.getScheduledState() == ScheduledState.DISABLED) {
throw new IllegalStateException(procNode + " is disabled, so it cannot be started");
}
StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider,
this.encryptor, getStateManager(procNode.getIdentifier()));
final ScheduleState scheduleState = getScheduleState(requireNonNull(procNode));
if (scheduleState.isScheduled()) {
return;
}
final int activeThreadCount = scheduleState.getActiveThreadCount();
if (activeThreadCount > 0) {
throw new IllegalStateException("Processor " + procNode.getName() + " cannot be started because it has " + activeThreadCount + " threads still running");
}
if (!procNode.isValid()) {
throw new IllegalStateException("Processor " + procNode.getName() + " is not in a valid state due to " + procNode.getValidationErrors());
}
final Runnable startProcRunnable = new Runnable() {
Runnable schedulingAgentCallback = new Runnable() {
@Override
@SuppressWarnings("deprecation")
public void run() {
try (final NarCloseable x = NarCloseable.withNarLoader()) {
final long lastStopTime = scheduleState.getLastStopTime();
final StandardProcessContext processContext = new StandardProcessContext(procNode, controllerServiceProvider, encryptor, getStateManager(procNode.getIdentifier()));
final Set<String> serviceIds = new HashSet<>();
for (final PropertyDescriptor descriptor : processContext.getProperties().keySet()) {
final Class<? extends ControllerService> serviceDefinition = descriptor.getControllerServiceDefinition();
if (serviceDefinition != null) {
final String serviceId = processContext.getProperty(descriptor).getValue();
if (serviceId != null) {
serviceIds.add(serviceId);
}
}
}
boolean needSleep = false;
attemptOnScheduled: while (true) {
try {
// We put this here so that we can sleep outside of the synchronized block, as
// we can't hold the synchronized block the whole time. If we do hold it the whole time,
// we will not be able to stop the controller service if it has trouble starting because
// the call to disable the service will block when attempting to synchronize on scheduleState.
if (needSleep) {
Thread.sleep(administrativeYieldMillis);
}
synchronized (scheduleState) {
for (final String serviceId : serviceIds) {
final boolean enabled = processContext.isControllerServiceEnabled(serviceId);
if (!enabled) {
LOG.debug("Controller Service with ID {} is not yet enabled, so will not start {} yet", serviceId, procNode);
needSleep = true;
continue attemptOnScheduled;
}
}
// if no longer scheduled to run, then we're finished. This can happen, for example,
// if the @OnScheduled method throws an Exception and the user stops the processor
// while we're administratively yielded.
// we also check if the schedule state's last start time is equal to what it was before.
// if not, then means that the processor has been stopped and started again, so we should just
// bail; another thread will be responsible for invoking the @OnScheduled methods.
if (!scheduleState.isScheduled() || scheduleState.getLastStopTime() != lastStopTime) {
return;
}
final SchedulingContext schedulingContext = new StandardSchedulingContext(processContext, controllerServiceProvider,
procNode, getStateManager(procNode.getIdentifier()));
ReflectionUtils.invokeMethodsWithAnnotations(OnScheduled.class, org.apache.nifi.processor.annotation.OnScheduled.class, procNode.getProcessor(), schedulingContext);
getSchedulingAgent(procNode).schedule(procNode, scheduleState);
heartbeater.heartbeat();
return;
}
} catch (final Exception e) {
final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e;
final ProcessorLog procLog = new SimpleProcessLogger(procNode.getIdentifier(), procNode.getProcessor());
procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {}",
new Object[]{procNode.getProcessor(), cause, administrativeYieldDuration}, cause);
LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause);
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, procNode.getProcessor(), processContext);
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, procNode.getProcessor(), processContext);
Thread.sleep(administrativeYieldMillis);
continue;
}
}
} catch (final Throwable t) {
final ProcessorLog procLog = new SimpleProcessLogger(procNode.getIdentifier(), procNode.getProcessor());
procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run", new Object[]{procNode.getProcessor(), t});
LOG.error("Failed to invoke @OnScheduled method due to {}", t.toString(), t);
}
getSchedulingAgent(procNode).schedule(procNode, scheduleState);
heartbeater.heartbeat();
}
};
procNode.start(this.componentLifeCycleThreadPool, this.administrativeYieldMillis, processContext, schedulingAgentCallback);
}
scheduleState.setScheduled(true);
procNode.setScheduledState(ScheduledState.RUNNING);
componentLifeCycleThreadPool.execute(startProcRunnable);
/**
* Stops the given {@link Processor} by invoking its
* {@link ProcessorNode#stop(ScheduledExecutorService, org.apache.nifi.processor.ProcessContext, Callable)}
* .
* @see StandardProcessorNode#stop(ScheduledExecutorService,
* org.apache.nifi.processor.ProcessContext, Callable)
*/
@Override
public synchronized void stopProcessor(final ProcessorNode procNode) {
StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider,
this.encryptor, getStateManager(procNode.getIdentifier()));
final ScheduleState state = getScheduleState(procNode);
procNode.stop(this.componentLifeCycleThreadPool, processContext, new Callable<Boolean>() {
@Override
public Boolean call() {
if (state.isScheduled()) {
getSchedulingAgent(procNode).unschedule(procNode, state);
}
return state.getActiveThreadCount() == 0;
}
});
}
@Override
public void yield(final ProcessorNode procNode) {
// This exists in the ProcessScheduler so that the scheduler can take advantage of the fact that
// the Processor was yielded and, as a result, avoid scheduling the Processor to potentially run
// (thereby skipping the overhead of the Context Switches) if nothing can be done.
// This exists in the ProcessScheduler so that the scheduler can take
// advantage of the fact that
// the Processor was yielded and, as a result, avoid scheduling the
// Processor to potentially run
// (thereby skipping the overhead of the Context Switches) if nothing
// can be done.
//
// We used to implement this feature by canceling all futures for the given Processor and
// re-submitting them with a delay. However, this became problematic, because we have situations where
// a Processor will wait several seconds (often 30 seconds in the case of a network timeout), and then yield
// the context. If this Processor has X number of threads, we end up submitting X new tasks while the previous
// X-1 tasks are still running. At this point, another thread could finish and do the same thing, resulting in
// We used to implement this feature by canceling all futures for the
// given Processor and
// re-submitting them with a delay. However, this became problematic,
// because we have situations where
// a Processor will wait several seconds (often 30 seconds in the case
// of a network timeout), and then yield
// the context. If this Processor has X number of threads, we end up
// submitting X new tasks while the previous
// X-1 tasks are still running. At this point, another thread could
// finish and do the same thing, resulting in
// an additional X-1 extra tasks being submitted.
//
// As a result, we simply removed this buggy implementation, as it was a very minor performance optimization
// As a result, we simply removed this buggy implementation, as it was a
// very minor performance optimization
// that gave very bad results.
}
/**
* Stops scheduling the given processor to run and invokes all methods on the underlying {@link org.apache.nifi.processor.Processor FlowFileProcessor} that are annotated with the
* {@link OnUnscheduled} annotation.
*/
@Override
public synchronized void stopProcessor(final ProcessorNode procNode) {
final ScheduleState state = getScheduleState(requireNonNull(procNode));
synchronized (state) {
if (!state.isScheduled()) {
procNode.setScheduledState(ScheduledState.STOPPED);
return;
}
state.setScheduled(false);
getSchedulingAgent(procNode).unschedule(procNode, state);
procNode.setScheduledState(ScheduledState.STOPPED);
}
final Runnable stopProcRunnable = new Runnable() {
@Override
public void run() {
try (final NarCloseable x = NarCloseable.withNarLoader()) {
final StandardProcessContext processContext = new StandardProcessContext(procNode, controllerServiceProvider, encryptor, getStateManager(procNode.getIdentifier()));
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, procNode.getProcessor(), processContext);
// If no threads currently running, call the OnStopped methods
if (state.getActiveThreadCount() == 0 && state.mustCallOnStoppedMethods()) {
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, procNode.getProcessor(), processContext);
heartbeater.heartbeat();
}
}
}
};
componentLifeCycleThreadPool.execute(stopProcRunnable);
}
@Override
public void registerEvent(final Connectable worker) {
getSchedulingAgent(worker).onEvent(worker);
@ -577,8 +476,6 @@ public final class StandardProcessScheduler implements ProcessScheduler {
if (procNode.getScheduledState() != ScheduledState.DISABLED) {
throw new IllegalStateException("Processor cannot be enabled because it is not disabled");
}
procNode.setScheduledState(ScheduledState.STOPPED);
}
@Override
@ -586,8 +483,6 @@ public final class StandardProcessScheduler implements ProcessScheduler {
if (procNode.getScheduledState() != ScheduledState.STOPPED) {
throw new IllegalStateException("Processor cannot be disabled because its state is set to " + procNode.getScheduledState());
}
procNode.setScheduledState(ScheduledState.DISABLED);
}
public synchronized void enableReportingTask(final ReportingTaskNode taskNode) {
@ -623,13 +518,10 @@ public final class StandardProcessScheduler implements ProcessScheduler {
* @return scheduled state
*/
private ScheduleState getScheduleState(final Object schedulable) {
ScheduleState scheduleState = scheduleStates.get(schedulable);
ScheduleState scheduleState = this.scheduleStates.get(schedulable);
if (scheduleState == null) {
scheduleState = new ScheduleState();
final ScheduleState previous = scheduleStates.putIfAbsent(schedulable, scheduleState);
if (previous != null) {
scheduleState = previous;
}
this.scheduleStates.putIfAbsent(schedulable, scheduleState);
}
return scheduleState;
}

10
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
View File

@ -42,7 +42,7 @@ import org.apache.nifi.util.NiFiProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TimerDrivenSchedulingAgent implements SchedulingAgent {
public class TimerDrivenSchedulingAgent extends AbstractSchedulingAgent {
private static final Logger logger = LoggerFactory.getLogger(TimerDrivenSchedulingAgent.class);
private final long noWorkYieldNanos;
@ -78,7 +78,7 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
public void schedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
public void doSchedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
final Runnable reportingTaskWrapper = new ReportingTaskWrapper(taskNode, scheduleState);
final long schedulingNanos = taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS);
@ -91,7 +91,7 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
public void schedule(final Connectable connectable, final ScheduleState scheduleState) {
public void doSchedule(final Connectable connectable, final ScheduleState scheduleState) {
final List<ScheduledFuture<?>> futures = new ArrayList<>();
for (int i = 0; i < connectable.getMaxConcurrentTasks(); i++) {
@ -197,7 +197,7 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
public void unschedule(final Connectable connectable, final ScheduleState scheduleState) {
public void doUnschedule(final Connectable connectable, final ScheduleState scheduleState) {
for (final ScheduledFuture<?> future : scheduleState.getFutures()) {
// stop scheduling to run but do not interrupt currently running tasks.
future.cancel(false);
@ -207,7 +207,7 @@ public class TimerDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
public void unschedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
public void doUnschedule(final ReportingTaskNode taskNode, final ScheduleState scheduleState) {
for (final ScheduledFuture<?> future : scheduleState.getFutures()) {
// stop scheduling to run but do not interrupt currently running tasks.
future.cancel(false);

1
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
View File

@ -457,6 +457,7 @@ public class StandardControllerServiceProvider implements ControllerServiceProvi
public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) {
final Set<String> identifiers = new HashSet<>();
for (final Map.Entry<String, ControllerServiceNode> entry : controllerServices.entrySet()) {
Class<? extends ControllerService> c = entry.getValue().getProxiedControllerService().getClass();
if (requireNonNull(serviceType).isAssignableFrom(entry.getValue().getProxiedControllerService().getClass())) {
identifiers.add(entry.getKey());
}

12
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
View File

@ -21,7 +21,6 @@ import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.repository.BatchingSessionFactory;
@ -39,7 +38,6 @@ import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.util.Connectables;
import org.apache.nifi.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -85,7 +83,6 @@ public class ContinuallyRunProcessorTask implements Callable<Boolean> {
}
@Override
@SuppressWarnings("deprecation")
public Boolean call() {
// make sure processor is not yielded
if (isYielded(procNode)) {
@ -191,15 +188,6 @@ public class ContinuallyRunProcessorTask implements Callable<Boolean> {
final long processingNanos = System.nanoTime() - startNanos;
// if the processor is no longer scheduled to run and this is the last thread,
// invoke the OnStopped methods
if (!scheduleState.isScheduled() && scheduleState.getActiveThreadCount() == 1 && scheduleState.mustCallOnStoppedMethods()) {
try (final NarCloseable x = NarCloseable.withNarLoader()) {
ReflectionUtils.quietlyInvokeMethodsWithAnnotations(OnStopped.class, org.apache.nifi.processor.annotation.OnStopped.class, procNode.getProcessor(), processContext);
flowController.heartbeat();
}
}
try {
final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(procNode.getIdentifier());
procEvent.setProcessingNanos(processingNanos);

760
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java Normal file
View File

@ -0,0 +1,760 @@
/*
* 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.controller.scheduling;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import java.io.File;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import org.apache.commons.io.FileUtils;
import org.apache.nifi.admin.service.AuditService;
import org.apache.nifi.admin.service.UserService;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.annotation.lifecycle.OnUnscheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.FlowController;
import org.apache.nifi.controller.ProcessScheduler;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.ScheduledState;
import org.apache.nifi.controller.repository.FlowFileEventRepository;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.provenance.MockProvenanceEventRepository;
import org.apache.nifi.util.NiFiProperties;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Validate Processor's life-cycle operation within the context of
* {@link FlowController} and {@link StandardProcessScheduler}
*/
public class TestProcessorLifecycle {
private static final Logger logger = LoggerFactory.getLogger(TestProcessorLifecycle.class);
@Before
public void before() {
System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties");
NiFiProperties.getInstance().setProperty(NiFiProperties.ADMINISTRATIVE_YIELD_DURATION, "1 sec");
NiFiProperties.getInstance().setProperty(NiFiProperties.STATE_MANAGEMENT_CONFIG_FILE, "target/test-classes/state-management.xml");
NiFiProperties.getInstance().setProperty(NiFiProperties.STATE_MANAGEMENT_LOCAL_PROVIDER_ID, "local-provider");
}
@After
public void after() throws Exception {
FileUtils.deleteDirectory(new File("./target/test-repo"));
FileUtils.deleteDirectory(new File("./target/content_repository"));
}
/**
* Will validate the idempotent nature of processor start operation which
* can be called multiple times without any side-effects.
*/
@Test
public void validateIdempotencyOfProcessorStartOperation() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
int randomDelayLimit = 3000;
this.randomOnTriggerDelay(testProcessor, randomDelayLimit);
final ProcessScheduler ps = fc.getProcessScheduler();
ExecutorService executor = Executors.newCachedThreadPool();
int startCallsCount = 100;
final CountDownLatch countDownCounter = new CountDownLatch(startCallsCount);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
for (int i = 0; i < startCallsCount; i++) {
executor.execute(new Runnable() {
@Override
public void run() {
ps.startProcessor(testProcNode);
countDownCounter.countDown();
}
});
}
assertTrue(countDownCounter.await(2000, TimeUnit.MILLISECONDS));
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
// regardless of how many threads attempted to start Processor, it must
// only be started once, hence have only single entry for @OnScheduled
assertEquals(1, testProcessor.operationNames.size());
assertEquals("@OnScheduled", testProcessor.operationNames.get(0));
fc.shutdown(true);
executor.shutdownNow();
}
/**
* Validates that stop calls are harmless and idempotent if processor is not
* in STARTING or RUNNING state.
*/
@Test
public void validateStopCallsAreMeaninglessIfProcessorNotStarted() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
// sets the scenario for the processor to run
int randomDelayLimit = 3000;
this.randomOnTriggerDelay(testProcessor, randomDelayLimit);
final ProcessScheduler ps = fc.getProcessScheduler();
ps.stopProcessor(testProcNode);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
assertTrue(testProcessor.operationNames.size() == 0);
fc.shutdown(true);
}
/**
* Validates the processors start/stop sequence where the order of
* operations can only be @OnScheduled, @OnUnscheduled, @OnStopped.
*/
@Test
public void validateSuccessfullAndOrderlyShutdown() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
int randomDelayLimit = 3000;
this.randomOnTriggerDelay(testProcessor, randomDelayLimit);
testProcNode.setMaxConcurrentTasks(4);
testProcNode.setScheduldingPeriod("500 millis");
testProcNode.setAutoTerminatedRelationships(Collections.singleton(new Relationship.Builder().name("success").build()));
testGroup.addProcessor(testProcNode);
fc.startProcessGroup(testGroup.getIdentifier());
Thread.sleep(2000); // let it run for a while
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
fc.stopAllProcessors();
Thread.sleep(randomDelayLimit); // up to randomDelayLimit, otherwise next assertion may fail as the processor still executing
// validates that regardless of how many running tasks, lifecycle
// operation are invoked atomically (once each).
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
// . . . hence only 3 operations must be in the list
assertEquals(3, testProcessor.operationNames.size());
// . . . and ordered as @OnScheduled, @OnUnscheduled, @OnStopped
assertEquals("@OnScheduled", testProcessor.operationNames.get(0));
assertEquals("@OnUnscheduled", testProcessor.operationNames.get(1));
assertEquals("@OnStopped", testProcessor.operationNames.get(2));
fc.shutdown(true);
}
/**
* Concurrency test that is basically hammers on both stop and start
* operation validating their idempotency.
*/
@Test
public void validateLifecycleOperationOrderWithConcurrentCallsToStartStop() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
final ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.noop(testProcessor);
final ProcessScheduler ps = fc.getProcessScheduler();
ExecutorService executor = Executors.newFixedThreadPool(100);
int startCallsCount = 10000;
final CountDownLatch countDownCounter = new CountDownLatch(startCallsCount);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
final Random random = new Random();
for (int i = 0; i < startCallsCount / 2; i++) {
executor.execute(new Runnable() {
@Override
public void run() {
LockSupport.parkNanos(random.nextInt(9000000));
ps.stopProcessor(testProcNode);
countDownCounter.countDown();
}
});
}
for (int i = 0; i < startCallsCount / 2; i++) {
executor.execute(new Runnable() {
@Override
public void run() {
LockSupport.parkNanos(random.nextInt(9000000));
ps.startProcessor(testProcNode);
countDownCounter.countDown();
}
});
}
assertTrue(countDownCounter.await(10000, TimeUnit.MILLISECONDS));
String previousOperation = null;
for (String operationName : testProcessor.operationNames) {
if (previousOperation == null || previousOperation.equals("@OnStopped")) {
assertEquals("@OnScheduled", operationName);
} else if (previousOperation.equals("@OnScheduled")) {
assertEquals("@OnUnscheduled", operationName);
} else if (previousOperation.equals("@OnUnscheduled")) {
assertTrue(operationName.equals("@OnStopped") || operationName.equals("@OnScheduled"));
}
previousOperation = operationName;
}
executor.shutdownNow();
fc.shutdown(true);
}
/**
* Validates that processor can be stopped before start sequence finished.
*/
@Test
public void validateProcessorUnscheduledAndStoppedWhenStopIsCalledBeforeProcessorFullyStarted() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
int delay = 2000;
this.longRunningOnSchedule(testProcessor, delay);
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
assertEquals(2, testProcessor.operationNames.size());
assertEquals("@OnScheduled", testProcessor.operationNames.get(0));
assertEquals("@OnUnscheduled", testProcessor.operationNames.get(1));
fc.shutdown(true);
}
/**
* Validates that Processor is eventually started once invocation
* of @OnSchedule stopped throwing exceptions.
*/
@Test
public void validateProcessScheduledAfterAdministrativeDelayDueToTheOnScheduledException() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.noop(testProcessor);
testProcessor.generateExceptionOnScheduled = true;
testProcessor.keepFailingOnScheduledTimes = 2;
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
}
/**
* Validates that Processor can be stopped when @OnScheduled constantly
* fails. Basically validates that the re-try loop breaks if user initiated
* stopProcessor.
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledConstantlyFails() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.longRunningOnUnschedule(testProcessor, 100);
testProcessor.generateExceptionOnScheduled = true;
testProcessor.keepFailingOnScheduledTimes = Integer.MAX_VALUE;
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
}
/**
* Validates that the Processor can be stopped when @OnScheduled blocks
* indefinitely but written to react to thread interrupts
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyInterruptable() throws Exception {
NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5000");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.blockingInterruptableOnUnschedule(testProcessor);
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
Thread.sleep(4000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
}
/**
* Validates that the Processor can be stopped when @OnScheduled blocks
* indefinitely and written to ignore thread interrupts
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyUninterruptable() throws Exception {
NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5000");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.blockingUninterruptableOnUnschedule(testProcessor);
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
Thread.sleep(4000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
}
/**
* Validates that processor can be stopped if onTrigger() keeps trowing
* exceptions.
*/
@Test
public void validateProcessorCanBeStoppedWhenOnTriggerThrowsException() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
// sets the scenario for the processor to run
this.noop(testProcessor);
testProcessor.generateExceptionOnTrigger = true;
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
}
/**
* Validate that processor will not be validated on failing
* PropertyDescriptor validation.
*/
@Test(expected = IllegalStateException.class)
public void validateStartFailsOnInvalidProcessorWithMissingProperty() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
ProcessScheduler ps = fc.getProcessScheduler();
try {
ps.startProcessor(testProcNode);
fail();
} finally {
fc.shutdown(true);
}
}
/**
* Validate that processor will not be validated on failing
* ControllerService validation (not enabled).
*/
@Test(expected = IllegalStateException.class)
public void validateStartFailsOnInvalidProcessorWithDisabledService() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ControllerServiceNode testServiceNode = fc.createControllerService(TestService.class.getName(), "serv", true);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
testProcNode.setProperty("S", testServiceNode.getIdentifier());
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
testProcessor.withService = true;
ProcessScheduler ps = fc.getProcessScheduler();
try {
ps.startProcessor(testProcNode);
fail();
} finally {
fc.shutdown(true);
}
}
/**
* The successful processor start with ControllerService dependency.
*/
@Test
public void validateStartSucceedsOnProcessorWithEnabledService() throws Exception {
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
ControllerServiceNode testServiceNode = fc.createControllerService(TestService.class.getName(), "foo", true);
ProcessorNode testProcNode = fc.createProcessor(TestProcessor.class.getName(), UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
testProcNode.setProperty("S", testServiceNode.getIdentifier());
TestProcessor testProcessor = (TestProcessor) testProcNode.getProcessor();
testProcessor.withService = true;
this.noop(testProcessor);
ProcessScheduler ps = fc.getProcessScheduler();
ps.enableControllerService(testServiceNode);
ps.startProcessor(testProcNode);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
fc.shutdown(true);
}
/**
* Scenario where onTrigger() is executed with random delay limited to
* 'delayLimit', yet with guaranteed exit from onTrigger().
*/
private void randomOnTriggerDelay(TestProcessor testProcessor, int delayLimit) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
RandomOrFixedDelayedRunnable delayedRunnable = new RandomOrFixedDelayedRunnable(delayLimit, true);
testProcessor.setScenario(emptyRunnable, emptyRunnable, emptyRunnable, delayedRunnable);
}
/**
* Scenario where @OnSchedule is executed with delay limited to
* 'delayLimit'.
*/
private void longRunningOnSchedule(TestProcessor testProcessor, int delayLimit) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
RandomOrFixedDelayedRunnable delayedRunnable = new RandomOrFixedDelayedRunnable(delayLimit, false);
testProcessor.setScenario(delayedRunnable, emptyRunnable, emptyRunnable, emptyRunnable);
}
/**
* Scenario where @OnUnschedule is executed with delay limited to
* 'delayLimit'.
*/
private void longRunningOnUnschedule(TestProcessor testProcessor, int delayLimit) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
RandomOrFixedDelayedRunnable delayedRunnable = new RandomOrFixedDelayedRunnable(delayLimit, false);
testProcessor.setScenario(emptyRunnable, delayedRunnable, emptyRunnable, emptyRunnable);
}
/**
* Scenario where @OnSchedule blocks indefinitely yet interruptible.
*/
private void blockingInterruptableOnUnschedule(TestProcessor testProcessor) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
BlockingInterruptableRunnable blockingRunnable = new BlockingInterruptableRunnable();
testProcessor.setScenario(blockingRunnable, emptyRunnable, emptyRunnable, emptyRunnable);
}
/**
* Scenario where @OnSchedule blocks indefinitely and un-interruptible.
*/
private void blockingUninterruptableOnUnschedule(TestProcessor testProcessor) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
BlockingUninterruptableRunnable blockingRunnable = new BlockingUninterruptableRunnable();
testProcessor.setScenario(blockingRunnable, emptyRunnable, emptyRunnable, emptyRunnable);
}
/**
* Scenario where all tasks are no op.
*/
private void noop(TestProcessor testProcessor) {
EmptyRunnable emptyRunnable = new EmptyRunnable();
testProcessor.setScenario(emptyRunnable, emptyRunnable, emptyRunnable, emptyRunnable);
}
/**
*
*/
private FlowController buildFlowControllerForTest() throws Exception {
NiFiProperties properties = NiFiProperties.getInstance();
properties.setProperty(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, MockProvenanceEventRepository.class.getName());
properties.setProperty("nifi.remote.input.socket.port", "");
properties.setProperty("nifi.remote.input.secure", "");
return FlowController.createStandaloneInstance(mock(FlowFileEventRepository.class), properties,
mock(UserService.class), mock(AuditService.class), null);
}
/**
*
*/
private void setControllerRootGroup(FlowController controller, ProcessGroup processGroup) {
try {
Method m = FlowController.class.getDeclaredMethod("setRootGroup", ProcessGroup.class);
m.setAccessible(true);
m.invoke(controller, processGroup);
controller.initializeFlow();
} catch (Exception e) {
throw new IllegalStateException("Failed to set root group", e);
}
}
/**
*/
public static class TestProcessor extends AbstractProcessor {
private Runnable onScheduleCallback;
private Runnable onUnscheduleCallback;
private Runnable onStopCallback;
private Runnable onTriggerCallback;
private boolean generateExceptionOnScheduled;
private boolean generateExceptionOnTrigger;
private boolean withService;
private int keepFailingOnScheduledTimes;
private int onScheduledExceptionCount;
private final List<String> operationNames = new LinkedList<>();
void setScenario(Runnable onScheduleCallback, Runnable onUnscheduleCallback, Runnable onStopCallback,
Runnable onTriggerCallback) {
this.onScheduleCallback = onScheduleCallback;
this.onUnscheduleCallback = onUnscheduleCallback;
this.onStopCallback = onStopCallback;
this.onTriggerCallback = onTriggerCallback;
}
@OnScheduled
public void schedule(ProcessContext ctx) {
this.operationNames.add("@OnScheduled");
if (this.generateExceptionOnScheduled
&& this.onScheduledExceptionCount++ < this.keepFailingOnScheduledTimes) {
throw new RuntimeException("Intentional");
}
this.onScheduleCallback.run();
}
@OnUnscheduled
public void unschedule() {
this.operationNames.add("@OnUnscheduled");
this.onUnscheduleCallback.run();
}
@OnStopped
public void stop() {
this.operationNames.add("@OnStopped");
this.onStopCallback.run();
}
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
PropertyDescriptor PROP = new PropertyDescriptor.Builder()
.name("P")
.description("Blah Blah")
.required(true)
.addValidator(new Validator() {
@Override
public ValidationResult validate(final String subject, final String value, final ValidationContext context) {
return new ValidationResult.Builder().subject(subject).input(value).valid(value != null && !value.isEmpty()).explanation(subject + " cannot be empty").build();
}
})
.build();
PropertyDescriptor SERVICE = new PropertyDescriptor.Builder()
.name("S")
.description("Blah Blah")
.required(true)
.identifiesControllerService(ITestservice.class)
.build();
return this.withService ? Arrays.asList(new PropertyDescriptor[] { PROP, SERVICE })
: Arrays.asList(new PropertyDescriptor[] { PROP });
}
@Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
if (this.generateExceptionOnTrigger) {
throw new RuntimeException("Intentional");
}
this.onTriggerCallback.run();
}
}
/**
*/
public static class TestService extends AbstractControllerService implements ITestservice {
}
/**
*/
public static interface ITestservice extends ControllerService {
}
/**
*/
private static class EmptyRunnable implements Runnable {
@Override
public void run() {
}
}
/**
*/
private static class BlockingInterruptableRunnable implements Runnable {
@Override
public void run() {
try {
Thread.sleep(Long.MAX_VALUE);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
/**
*/
private static class BlockingUninterruptableRunnable implements Runnable {
@Override
public void run() {
while (true) {
try {
Thread.sleep(Long.MAX_VALUE);
} catch (InterruptedException e) {
// ignore
}
}
}
}
/**
*/
private static class RandomOrFixedDelayedRunnable implements Runnable {
private final int delayLimit;
private final boolean randomDelay;
public RandomOrFixedDelayedRunnable(int delayLimit, boolean randomDelay) {
this.delayLimit = delayLimit;
this.randomDelay = randomDelay;
}
Random random = new Random();
@Override
public void run() {
try {
if (this.randomDelay) {
Thread.sleep(random.nextInt(this.delayLimit));
} else {
Thread.sleep(this.delayLimit);
}
} catch (InterruptedException e) {
logger.warn("Interrupted while sleeping");
Thread.currentThread().interrupt();
}
}
}
}

19
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
View File

@ -45,6 +45,7 @@ import org.apache.nifi.processor.StandardProcessorInitializationContext;
import org.apache.nifi.processor.StandardValidationContextFactory;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@ -190,7 +191,9 @@ public class TestStandardControllerServiceProvider {
}
}
@Test(timeout=10000)
@Test(timeout = 10000)
@Ignore // this may be obsolete since TestProcessorLifecycle covers this
// scenario without mocks
public void testStartStopReferencingComponents() {
final ProcessScheduler scheduler = createScheduler();
final StandardControllerServiceProvider provider = new StandardControllerServiceProvider(scheduler, null, stateManagerProvider);
@ -218,7 +221,7 @@ public class TestStandardControllerServiceProvider {
public Object answer(InvocationOnMock invocation) throws Throwable {
final ProcessorNode procNode = (ProcessorNode) invocation.getArguments()[0];
procNode.verifyCanStart();
procNode.setScheduledState(ScheduledState.RUNNING);
// procNode.setScheduledState(ScheduledState.RUNNING);
return null;
}
}).when(mockProcessGroup).startProcessor(Mockito.any(ProcessorNode.class));
@ -228,7 +231,7 @@ public class TestStandardControllerServiceProvider {
public Object answer(final InvocationOnMock invocation) throws Throwable {
final ProcessorNode procNode = (ProcessorNode) invocation.getArguments()[0];
procNode.verifyCanStop();
procNode.setScheduledState(ScheduledState.STOPPED);
// procNode.setScheduledState(ScheduledState.STOPPED);
return null;
}
}).when(mockProcessGroup).stopProcessor(Mockito.any(ProcessorNode.class));
@ -466,16 +469,16 @@ public class TestStandardControllerServiceProvider {
final ProcessorNode procNode = createProcessor(scheduler, provider);
serviceNode.addReference(procNode);
procNode.setScheduledState(ScheduledState.STOPPED);
// procNode.setScheduledState(ScheduledState.STOPPED);
provider.unscheduleReferencingComponents(serviceNode);
assertEquals(ScheduledState.STOPPED, procNode.getScheduledState());
procNode.setScheduledState(ScheduledState.RUNNING);
// procNode.setScheduledState(ScheduledState.RUNNING);
provider.unscheduleReferencingComponents(serviceNode);
assertEquals(ScheduledState.STOPPED, procNode.getScheduledState());
procNode.setScheduledState(ScheduledState.DISABLED);
provider.unscheduleReferencingComponents(serviceNode);
assertEquals(ScheduledState.DISABLED, procNode.getScheduledState());
// procNode.setScheduledState(ScheduledState.DISABLED);
// provider.unscheduleReferencingComponents(serviceNode);
// assertEquals(ScheduledState.DISABLED, procNode.getScheduledState());
}
}

38
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/state-management.xml Normal file
View File

@ -0,0 +1,38 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<!--
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.
-->
<!--
This file lists the authority providers to use when running securely. In order
to use a specific provider it must be configured here and it's identifier
must be specified in the nifi.properties file.
-->
<stateManagement>
<!--
This file provides a mechanism for defining and configuring the State Providers
that should be used for storing state locally and across a NiFi cluster.
-->
<!--
State Provider that stores state locally in a configurable directory. This Provider requires the following properties:
Directory - the directory to store components' state in. If the directory being used is a sub-directory of the NiFi installation, it
is important that the directory be copied over to the new version when upgrading NiFi.
-->
<local-provider>
<id>local-provider</id>
<class>org.apache.nifi.controller.state.providers.local.WriteAheadLocalStateProvider</class>
<property name="Directory">target/test-classes/access-control/state-management</property>
</local-provider>
</stateManagement>