Return-Path: 
 <commits-return-21762-apmail-nifi-commits-archive=nifi.apache.org@nifi.apache.org>
X-Original-To: apmail-nifi-commits-archive@minotaur.apache.org
Delivered-To: apmail-nifi-commits-archive@minotaur.apache.org
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by minotaur.apache.org (Postfix) with SMTP id 5D01810900
	for <apmail-nifi-commits-archive@minotaur.apache.org>;
 Fri, 11 Mar 2016 17:55:33 +0000 (UTC)
Received: (qmail 82023 invoked by uid 500); 11 Mar 2016 17:55:33 -0000
Delivered-To: apmail-nifi-commits-archive@nifi.apache.org
Received: (qmail 81927 invoked by uid 500); 11 Mar 2016 17:55:33 -0000
Mailing-List: contact commits-help@nifi.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:commits-help@nifi.apache.org>
List-Unsubscribe: <mailto:commits-unsubscribe@nifi.apache.org>
List-Post: <mailto:commits@nifi.apache.org>
List-Id: <commits.nifi.apache.org>
Reply-To: dev@nifi.apache.org
Delivered-To: mailing list commits@nifi.apache.org
Received: (qmail 81784 invoked by uid 99); 11 Mar 2016 17:55:32 -0000
Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org)
 (140.211.11.23)
    by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 11 Mar 2016 17:55:32 +0000
Received: by git1-us-west.apache.org (ASF Mail Server at
 git1-us-west.apache.org, from userid 33)
	id 85102E00A1; Fri, 11 Mar 2016 17:55:32 +0000 (UTC)
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: markap14@apache.org
To: commits@nifi.apache.org
Date: Fri, 11 Mar 2016 17:55:35 -0000
Message-Id: <43044796153e44c994df7beb16f64c98@git.apache.org>
In-Reply-To: <0b30801791a24da991a6da340b10247b@git.apache.org>
References: <0b30801791a24da991a6da340b10247b@git.apache.org>
X-Mailer: ASF-Git Admin Mailer
Subject: [4/9] nifi git commit: 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
 blo

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
index 3fa6b52..31c1ca4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/QuartzSchedulingAgent.java
@@ -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);
     }
 

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java
index e03cc05..de2c35a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java
@@ -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

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
index ef27fb5..84d667f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java
@@ -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,171 +284,74 @@ 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();
             }
         };
-
-        scheduleState.setScheduled(true);
-        procNode.setScheduledState(ScheduledState.RUNNING);
-
-        componentLifeCycleThreadPool.execute(startProcRunnable);
-    }
-
-    @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.
-        //
-        // 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
-        // that gave very bad results.
+        procNode.start(this.componentLifeCycleThreadPool, this.administrativeYieldMillis, processContext, schedulingAgentCallback);
     }
 
     /**
-     * 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.
+     * 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) {
-        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() {
+        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 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();
-                    }
+            public Boolean call() {
+                if (state.isScheduled()) {
+                    getSchedulingAgent(procNode).unschedule(procNode, state);
                 }
+                return state.getActiveThreadCount() == 0;
             }
-        };
+        });
+    }
 
-        componentLifeCycleThreadPool.execute(stopProcRunnable);
+    @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.
+        //
+        // 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
+        // that gave very bad results.
     }
 
     @Override
@@ -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;
     }

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
index 04db549..76c413f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java
@@ -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);

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
index 77dc87e..3b9b073 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceProvider.java
@@ -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());
             }

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
index de91a6d..91c17ae 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ContinuallyRunProcessorTask.java
@@ -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);

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java
new file mode 100644
index 0000000..2cce14d
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestProcessorLifecycle.java
@@ -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();
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
index 0dcacb5..9b35238 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/service/TestStandardControllerServiceProvider.java
@@ -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());
     }
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/state-management.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/state-management.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/state-management.xml
new file mode 100644
index 0000000..1714c7d
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/state-management.xml
@@ -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>
\ No newline at end of file