Repository: nifi Updated Branches: refs/heads/master 5b58e5a25 -> 91a59a8a5
NIFI-2545: Ensure that when @OnUnscheduled and @OnStopped methods are called that the active thread count takes that thread into account This closes #836. Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/91a59a8a Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/91a59a8a Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/91a59a8a Branch: refs/heads/master Commit: 91a59a8a5af72d4e8b92476ffeb679b189e9542a Parents: 5b58e5a Author: Mark Payne <[email protected]> Authored: Thu Aug 11 11:18:50 2016 -0400 Committer: Mark Payne <[email protected]> Committed: Thu Aug 18 08:33:45 2016 -0400 ---------------------------------------------------------------------- .../nifi/annotation/lifecycle/OnScheduled.java | 11 ++ .../nifi/annotation/lifecycle/OnStopped.java | 9 ++ .../annotation/lifecycle/OnUnscheduled.java | 10 ++ .../apache/nifi/controller/ProcessorNode.java | 29 ++-- .../controller/scheduling/ScheduleState.java | 102 +++++++++++++ .../controller/scheduling/SchedulingAgent.java | 45 ++++++ .../nifi/controller/StandardProcessorNode.java | 51 ++++--- .../controller/scheduling/ScheduleState.java | 102 ------------- .../controller/scheduling/SchedulingAgent.java | 45 ------ .../scheduling/StandardProcessScheduler.java | 22 +-- .../nifi/processors/standard/DebugFlow.java | 149 +++++++++++++------ .../nifi/processors/standard/TestDebugFlow.java | 7 +- 12 files changed, 337 insertions(+), 245 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java index 8fa752b..b320858 100644 --- a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java +++ b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java @@ -53,11 +53,22 @@ import java.lang.annotation.Target; * {@link org.apache.nifi.controller.ConfigurationContext ConfigurationContext}. * </p> * + * <p> * If any method annotated with this annotation throws any Throwable, the * framework will wait a while and then attempt to invoke the method again. This * will continue until the method succeeds, and the component will then be * scheduled to run after this method return successfully. + * </p> * + * <p><b>Implementation Guidelines:</b> + * <ul> + * <li>Methods with this annotation are expected to perform very quick, short-lived tasks. If the function is + * expensive or long-lived, the logic should be performed in the {@code onTrigger} method instead.</li> + * <li>If a method with this annotation does not return (exceptionally or otherwise) within a short period + * of time (the duration is configurable in the properties file), the Thread may be interrupted.</li> + * <li>Methods that make use of this interface should honor Java's Thread interruption mechanisms and not swallow + * {@link InterruptedException}.</li> + * </ul> */ @Documented @Target({ElementType.METHOD}) http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnStopped.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnStopped.java b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnStopped.java index 622f158..cdec8d0 100644 --- a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnStopped.java +++ b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnStopped.java @@ -57,6 +57,15 @@ import org.apache.nifi.processor.ProcessContext; * component is a Processor. * </p> * + * <p><b>Implementation Guidelines:</b> + * <ul> + * <li>Methods with this annotation are expected to perform very quick, short-lived tasks. If the function is + * expensive or long-lived, the logic should be performed in the {@code onTrigger} method instead.</li> + * <li>If a method with this annotation does not return (exceptionally or otherwise) within a short period + * of time (the duration is configurable in the properties file), the Thread may be interrupted.</li> + * <li>Methods that make use of this interface should honor Java's Thread interruption mechanisms and not swallow + * {@link InterruptedException}.</li> + * </ul> */ @Documented @Target({ElementType.METHOD}) http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java index b7d5027..ea041ae 100644 --- a/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java +++ b/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java @@ -50,6 +50,16 @@ import java.lang.annotation.Target; * Task, that argument must be of type * {@link org.apache.nifi.controller.ConfigurationContext ConfigurationContext}. * </p> + * + * <p><b>Implementation Guidelines:</b> + * <ul> + * <li>Methods with this annotation are expected to perform very quick, short-lived tasks. If the function is + * expensive or long-lived, the logic should be performed in the {@code onTrigger} method instead.</li> + * <li>If a method with this annotation does not return (exceptionally or otherwise) within a short period + * of time (the duration is configurable in the properties file), the Thread may be interrupted.</li> + * <li>Methods that make use of this interface should honor Java's Thread interruption mechanisms and not swallow + * {@link InterruptedException}.</li> + * </ul> */ @Documented @Target({ElementType.METHOD}) http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java index 29c2cef..0fe306c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/ProcessorNode.java @@ -16,8 +16,16 @@ */ package org.apache.nifi.controller; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.controller.scheduling.ScheduleState; +import org.apache.nifi.controller.scheduling.SchedulingAgent; import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.logging.LogLevel; @@ -28,13 +36,6 @@ import org.apache.nifi.scheduling.SchedulingStrategy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.Callable; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - public abstract class ProcessorNode extends AbstractConfiguredComponent implements Connectable { private static final Logger logger = LoggerFactory.getLogger(ProcessorNode.class); @@ -169,16 +170,14 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen * @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. + * @param schedulingAgent + * the SchedulingAgent that is responsible for managing the scheduling of the ProcessorNode + * @param scheduleState + * the ScheduleState that can be used to ensure that the running state (STOPPED, RUNNING, etc.) + * as well as the active thread counts are kept in sync */ public abstract <T extends ProcessContext & ControllerServiceLookup> void stop(ScheduledExecutorService scheduler, - T processContext, Callable<Boolean> activeThreadMonitorCallback); + T processContext, SchedulingAgent schedulingAgent, ScheduleState scheduleState); /** * Will set the state of the processor to STOPPED which essentially implies http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java new file mode 100644 index 0000000..de2c35a --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java @@ -0,0 +1,102 @@ +/* + * 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 java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.nifi.annotation.lifecycle.OnStopped; + +public class ScheduleState { + + private final AtomicInteger activeThreadCount = new AtomicInteger(0); + private final AtomicBoolean scheduled = new AtomicBoolean(false); + private final Set<ScheduledFuture<?>> futures = new HashSet<>(); + private final AtomicBoolean mustCallOnStoppedMethods = new AtomicBoolean(false); + private volatile long lastStopTime = -1; + + public int incrementActiveThreadCount() { + return activeThreadCount.incrementAndGet(); + } + + public int decrementActiveThreadCount() { + return activeThreadCount.decrementAndGet(); + } + + public int getActiveThreadCount() { + return activeThreadCount.get(); + } + + public boolean isScheduled() { + return scheduled.get(); + } + + void setScheduled(final boolean scheduled) { + this.scheduled.set(scheduled); + mustCallOnStoppedMethods.set(true); + + if (!scheduled) { + lastStopTime = System.currentTimeMillis(); + } + } + + public long getLastStopTime() { + 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 + * {@link OnStopped @OnStopped} + * + * @return <code>true</code> if the caller is required to call Processor methods annotated with + * @OnStopped, <code>false</code> otherwise + */ + public boolean mustCallOnStoppedMethods() { + return mustCallOnStoppedMethods.getAndSet(false); + } + + /** + * Establishes the list of relevant futures for this processor. Replaces any previously held futures. + * + * @param newFutures futures + */ + public synchronized void setFutures(final Collection<ScheduledFuture<?>> newFutures) { + futures.clear(); + futures.addAll(newFutures); + } + + public synchronized void replaceFuture(final ScheduledFuture<?> oldFuture, final ScheduledFuture<?> newFuture) { + futures.remove(oldFuture); + futures.add(newFuture); + } + + public synchronized Set<ScheduledFuture<?>> getFutures() { + return Collections.unmodifiableSet(futures); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java new file mode 100644 index 0000000..c48e13f --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java @@ -0,0 +1,45 @@ +/* + * 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 java.util.concurrent.TimeUnit; + +import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.controller.ReportingTaskNode; + +public interface SchedulingAgent { + + void schedule(Connectable connectable, ScheduleState scheduleState); + + void unschedule(Connectable connectable, ScheduleState scheduleState); + + void onEvent(Connectable connectable); + + void schedule(ReportingTaskNode taskNode, ScheduleState scheduleState); + + void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState); + + void setMaxThreadCount(int maxThreads); + + void setAdministrativeYieldDuration(String duration); + + String getAdministrativeYieldDuration(); + + long getAdministrativeYieldDuration(TimeUnit timeUnit); + + void shutdown(); +} http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java index 5ff97ef..cb12ab0 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java @@ -40,6 +40,8 @@ import org.apache.nifi.connectable.Connectable; import org.apache.nifi.connectable.ConnectableType; import org.apache.nifi.connectable.Connection; import org.apache.nifi.connectable.Position; +import org.apache.nifi.controller.scheduling.ScheduleState; +import org.apache.nifi.controller.scheduling.SchedulingAgent; import org.apache.nifi.controller.service.ControllerServiceNode; import org.apache.nifi.controller.service.ControllerServiceProvider; import org.apache.nifi.groups.ProcessGroup; @@ -1255,8 +1257,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; - procLog.error( "{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {}", - new Object[] { StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis + " milliseconds" }, cause); + procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", + new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); @@ -1285,17 +1287,18 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable /** * Will idempotently stop the processor using the following sequence: <i> * <ul> - * <li>Transition (atomically) Processor's scheduled state form RUNNING to - * STOPPING. If the above state transition succeeds, then execute the stop - * task (asynchronously) where 'activeThreadMonitorCallback' provided by the - * {@link ProcessScheduler} will be called to check if this processor still - * has active threads. If it does, the task will be re-scheduled with delay - * of 100 milliseconds until there are no more active threads, at which - * point processor's @OnUnscheduled and @OnStopped operation will be invoked + * <li>Transition (atomically) Processor's scheduled state from RUNNING to + * STOPPING. If the above state transition succeeds, then invoke any method + * on the Processor with the {@link OnUnscheduled} annotation. Once those methods + * have been called and returned (either normally or exceptionally), start checking + * to see if all of the Processor's active threads have finished. If not, check again + * every 100 milliseconds until they have. + * Once all after threads have completed, the processor's @OnStopped operation will be invoked * and its scheduled state is set to STOPPED which completes processor stop * sequence.</li> * </ul> * </i> + * * <p> * If for some reason processor's scheduled state can not be transitioned to * STOPPING (e.g., the processor didn't finish @OnScheduled operation when @@ -1309,26 +1312,36 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable */ @Override public <T extends ProcessContext & ControllerServiceLookup> void stop(final ScheduledExecutorService scheduler, - final T processContext, final Callable<Boolean> activeThreadMonitorCallback) { + final T processContext, final SchedulingAgent schedulingAgent, final ScheduleState scheduleState) { LOG.info("Stopping processor: " + this.processor.getClass()); if (this.scheduledState.compareAndSet(ScheduledState.RUNNING, ScheduledState.STOPPING)) { // will ensure that the Processor represented by this node can only be stopped once - // will continue to monitor active threads, invoking OnStopped once - // there are none + scheduleState.incrementActiveThreadCount(); + + // will continue to monitor active threads, invoking OnStopped once there are no + // active threads (with the exception of the thread performing shutdown operations) scheduler.execute(new Runnable() { - boolean unscheduled = false; @Override public void run() { - if (!this.unscheduled){ - ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); - this.unscheduled = true; - } try { - if (activeThreadMonitorCallback.call()) { + if (scheduleState.isScheduled()) { + schedulingAgent.unschedule(StandardProcessorNode.this, scheduleState); + try (final NarCloseable nc = NarCloseable.withNarLoader()) { + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); + } + } + + // all threads are complete if the active thread count is 1. This is because this thread that is + // performing the lifecycle actions counts as 1 thread. + final boolean allThreadsComplete = scheduleState.getActiveThreadCount() == 1; + if (allThreadsComplete) { try (final NarCloseable nc = NarCloseable.withNarLoader()) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); } + + scheduleState.decrementActiveThreadCount(); scheduledState.set(ScheduledState.STOPPED); } else { + // Not all of the active threads have finished. Try again in 100 milliseconds. scheduler.schedule(this, 100, TimeUnit.MILLISECONDS); } } catch (final Exception e) { @@ -1393,7 +1406,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable + this.processor.getClass().getSimpleName() + "' processor to finish. An attempt is made to cancel the task via Thread.interrupt(). However it does not " + "guarantee that the task will be canceled since the code inside current OnScheduled operation may " - + "have been written to ignore interrupts which may result in runaway thread which could lead to more issues " + + "have been written to ignore interrupts which may result in a runaway thread. This could lead to more issues, " + "eventually requiring NiFi to be restarted. This is usually a bug in the target Processor '" + this.processor + "' that needs to be documented, reported and eventually fixed."); throw new RuntimeException("Timed out while executing one of processor's OnScheduled task.", e); http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/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 deleted file mode 100644 index de2c35a..0000000 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/ScheduleState.java +++ /dev/null @@ -1,102 +0,0 @@ -/* - * 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 java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.Set; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; - -import org.apache.nifi.annotation.lifecycle.OnStopped; - -public class ScheduleState { - - private final AtomicInteger activeThreadCount = new AtomicInteger(0); - private final AtomicBoolean scheduled = new AtomicBoolean(false); - private final Set<ScheduledFuture<?>> futures = new HashSet<>(); - private final AtomicBoolean mustCallOnStoppedMethods = new AtomicBoolean(false); - private volatile long lastStopTime = -1; - - public int incrementActiveThreadCount() { - return activeThreadCount.incrementAndGet(); - } - - public int decrementActiveThreadCount() { - return activeThreadCount.decrementAndGet(); - } - - public int getActiveThreadCount() { - return activeThreadCount.get(); - } - - public boolean isScheduled() { - return scheduled.get(); - } - - void setScheduled(final boolean scheduled) { - this.scheduled.set(scheduled); - mustCallOnStoppedMethods.set(true); - - if (!scheduled) { - lastStopTime = System.currentTimeMillis(); - } - } - - public long getLastStopTime() { - 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 - * {@link OnStopped @OnStopped} - * - * @return <code>true</code> if the caller is required to call Processor methods annotated with - * @OnStopped, <code>false</code> otherwise - */ - public boolean mustCallOnStoppedMethods() { - return mustCallOnStoppedMethods.getAndSet(false); - } - - /** - * Establishes the list of relevant futures for this processor. Replaces any previously held futures. - * - * @param newFutures futures - */ - public synchronized void setFutures(final Collection<ScheduledFuture<?>> newFutures) { - futures.clear(); - futures.addAll(newFutures); - } - - public synchronized void replaceFuture(final ScheduledFuture<?> oldFuture, final ScheduledFuture<?> newFuture) { - futures.remove(oldFuture); - futures.add(newFuture); - } - - public synchronized Set<ScheduledFuture<?>> getFutures() { - return Collections.unmodifiableSet(futures); - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java deleted file mode 100644 index c48e13f..0000000 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/SchedulingAgent.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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 java.util.concurrent.TimeUnit; - -import org.apache.nifi.connectable.Connectable; -import org.apache.nifi.controller.ReportingTaskNode; - -public interface SchedulingAgent { - - void schedule(Connectable connectable, ScheduleState scheduleState); - - void unschedule(Connectable connectable, ScheduleState scheduleState); - - void onEvent(Connectable connectable); - - void schedule(ReportingTaskNode taskNode, ScheduleState scheduleState); - - void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState); - - void setMaxThreadCount(int maxThreads); - - void setAdministrativeYieldDuration(String duration); - - String getAdministrativeYieldDuration(); - - long getAdministrativeYieldDuration(TimeUnit timeUnit); - - void shutdown(); -} http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/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 10a746e..0a76e4f 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 @@ -295,10 +295,9 @@ public final class StandardProcessScheduler implements ProcessScheduler { /** * Starts the given {@link Processor} by invoking its * {@link ProcessorNode#start(ScheduledExecutorService, long, org.apache.nifi.processor.ProcessContext, Runnable)} - * . + * method. * - * @see StandardProcessorNode#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) { @@ -329,11 +328,10 @@ public final class StandardProcessScheduler implements ProcessScheduler { /** * Stops the given {@link Processor} by invoking its - * {@link ProcessorNode#stop(ScheduledExecutorService, org.apache.nifi.processor.ProcessContext, Callable)} - * . + * {@link ProcessorNode#stop(ScheduledExecutorService, org.apache.nifi.processor.ProcessContext, SchedulingAgent, ScheduleState)} + * method. * - * @see StandardProcessorNode#stop(ScheduledExecutorService, - * org.apache.nifi.processor.ProcessContext, Callable) + * @see StandardProcessorNode#stop(ScheduledExecutorService, org.apache.nifi.processor.ProcessContext, SchedulingAgent, ScheduleState) */ @Override public synchronized void stopProcessor(final ProcessorNode procNode) { @@ -341,15 +339,7 @@ public final class StandardProcessScheduler implements ProcessScheduler { this.encryptor, getStateManager(procNode.getIdentifier()), variableRegistry); 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; - } - }); + procNode.stop(this.componentLifeCycleThreadPool, processContext, getSchedulingAgent(procNode), state); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DebugFlow.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DebugFlow.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DebugFlow.java index cd5ce54..3d12ee5 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DebugFlow.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/DebugFlow.java @@ -25,6 +25,7 @@ import java.util.HashSet; import java.util.List; import java.util.Random; import java.util.Set; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import org.apache.http.annotation.ThreadSafe; @@ -39,6 +40,8 @@ import org.apache.nifi.components.Validator; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.annotation.lifecycle.OnUnscheduled; import org.apache.nifi.processor.AbstractProcessor; import org.apache.nifi.processor.DataUnit; import org.apache.nifi.processor.ProcessContext; @@ -117,19 +120,7 @@ public class DebugFlow extends AbstractProcessor { .description("Exception class to be thrown (must extend java.lang.RuntimeException).") .required(true) .defaultValue("java.lang.RuntimeException") - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .addValidator(new Validator() { - @Override - public ValidationResult validate(String subject, String input, ValidationContext context) { - Class<? extends RuntimeException> klass = classNameToRuntimeExceptionClass(input); - return new ValidationResult.Builder() - .subject(subject) - .input(input) - .valid(klass != null && (RuntimeException.class.isAssignableFrom(klass))) - .explanation(subject + " class must exist and extend java.lang.RuntimeException") - .build(); - } - }) + .addValidator(new RuntimeExceptionValidator()) .build(); static final PropertyDescriptor NO_FF_SKIP_ITERATIONS = new PropertyDescriptor.Builder() @@ -158,19 +149,7 @@ public class DebugFlow extends AbstractProcessor { .description("Exception class to be thrown if no FlowFile (must extend java.lang.RuntimeException).") .required(true) .defaultValue("java.lang.RuntimeException") - .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) - .addValidator(new Validator() { - @Override - public ValidationResult validate(String subject, String input, ValidationContext context) { - Class<? extends RuntimeException> klass = classNameToRuntimeExceptionClass(input); - return new ValidationResult.Builder() - .subject(subject) - .input(input) - .valid(klass != null && (RuntimeException.class.isAssignableFrom(klass))) - .explanation(subject + " class must exist and extend java.lang.RuntimeException") - .build(); - } - }) + .addValidator(new RuntimeExceptionValidator()) .build(); static final PropertyDescriptor WRITE_ITERATIONS = new PropertyDescriptor.Builder() .name("Write Iterations") @@ -187,6 +166,50 @@ public class DebugFlow extends AbstractProcessor { .defaultValue("1 KB") .build(); + static final PropertyDescriptor ON_SCHEDULED_SLEEP_TIME = new PropertyDescriptor.Builder() + .name("@OnScheduled Pause Time") + .description("Specifies how long the processor should sleep in the @OnScheduled method, so that the processor can be forced to take a long time to start up") + .required(true) + .defaultValue("0 sec") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + static final PropertyDescriptor ON_SCHEDULED_FAIL = new PropertyDescriptor.Builder() + .name("Fail When @OnScheduled called") + .description("Specifies whether or not the Processor should throw an Exception when the methods annotated with @OnScheduled are called") + .required(true) + .allowableValues("true", "false") + .defaultValue("false") + .build(); + static final PropertyDescriptor ON_UNSCHEDULED_SLEEP_TIME = new PropertyDescriptor.Builder() + .name("@OnUnscheduled Pause Time") + .description("Specifies how long the processor should sleep in the @OnUnscheduled method, so that the processor can be forced to take a long time to respond when user clicks stop") + .required(true) + .defaultValue("0 sec") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + static final PropertyDescriptor ON_UNSCHEDULED_FAIL = new PropertyDescriptor.Builder() + .name("Fail When @OnUnscheduled called") + .description("Specifies whether or not the Processor should throw an Exception when the methods annotated with @OnUnscheduled are called") + .required(true) + .allowableValues("true", "false") + .defaultValue("false") + .build(); + static final PropertyDescriptor ON_STOPPED_SLEEP_TIME = new PropertyDescriptor.Builder() + .name("@OnStopped Pause Time") + .description("Specifies how long the processor should sleep in the @OnStopped method, so that the processor can be forced to take a long time to shutdown") + .required(true) + .defaultValue("0 sec") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .build(); + static final PropertyDescriptor ON_STOPPED_FAIL = new PropertyDescriptor.Builder() + .name("Fail When @OnStopped called") + .description("Specifies whether or not the Processor should throw an Exception when the methods annotated with @OnStopped are called") + .required(true) + .allowableValues("true", "false") + .defaultValue("false") + .build(); + + private volatile Integer flowFileMaxSuccess = 0; private volatile Integer flowFileMaxFailure = 0; private volatile Integer flowFileMaxRollback = 0; @@ -246,6 +269,13 @@ public class DebugFlow extends AbstractProcessor { propList.add(NO_FF_EXCEPTION_CLASS); propList.add(WRITE_ITERATIONS); propList.add(CONTENT_SIZE); + propList.add(ON_SCHEDULED_SLEEP_TIME); + propList.add(ON_SCHEDULED_FAIL); + propList.add(ON_UNSCHEDULED_SLEEP_TIME); + propList.add(ON_UNSCHEDULED_FAIL); + propList.add(ON_STOPPED_SLEEP_TIME); + propList.add(ON_STOPPED_FAIL); + propertyDescriptors.compareAndSet(null, Collections.unmodifiableList(propList)); } return propertyDescriptors.get(); @@ -253,7 +283,8 @@ public class DebugFlow extends AbstractProcessor { } @OnScheduled - public void onScheduled(ProcessContext context) { + @SuppressWarnings("unchecked") + public void onScheduled(ProcessContext context) throws ClassNotFoundException, InterruptedException { flowFileMaxSuccess = context.getProperty(FF_SUCCESS_ITERATIONS).asInteger(); flowFileMaxFailure = context.getProperty(FF_FAILURE_ITERATIONS).asInteger(); flowFileMaxYield = context.getProperty(FF_ROLLBACK_YIELD_ITERATIONS).asInteger(); @@ -265,10 +296,38 @@ public class DebugFlow extends AbstractProcessor { noFlowFileMaxSkip = context.getProperty(NO_FF_SKIP_ITERATIONS).asInteger(); curr_ff_resp.reset(); curr_noff_resp.reset(); - flowFileExceptionClass = classNameToRuntimeExceptionClass(context.getProperty(FF_EXCEPTION_CLASS).toString()); - noFlowFileExceptionClass = classNameToRuntimeExceptionClass(context.getProperty(NO_FF_EXCEPTION_CLASS).toString()); + flowFileExceptionClass = (Class<? extends RuntimeException>) Class.forName(context.getProperty(FF_EXCEPTION_CLASS).toString()); + noFlowFileExceptionClass = (Class<? extends RuntimeException>) Class.forName(context.getProperty(NO_FF_EXCEPTION_CLASS).toString()); + + sleep(context.getProperty(ON_SCHEDULED_SLEEP_TIME).asTimePeriod(TimeUnit.MILLISECONDS)); + fail(context.getProperty(ON_SCHEDULED_FAIL).asBoolean(), OnScheduled.class); + } + + @OnUnscheduled + public void onUnscheduled(final ProcessContext context) throws InterruptedException { + sleep(context.getProperty(ON_UNSCHEDULED_SLEEP_TIME).asTimePeriod(TimeUnit.MILLISECONDS)); + fail(context.getProperty(ON_UNSCHEDULED_FAIL).asBoolean(), OnUnscheduled.class); + } + + @OnStopped + public void onStopped(final ProcessContext context) throws InterruptedException { + sleep(context.getProperty(ON_STOPPED_SLEEP_TIME).asTimePeriod(TimeUnit.MILLISECONDS)); + fail(context.getProperty(ON_STOPPED_FAIL).asBoolean(), OnStopped.class); + } + + private void sleep(final long millis) throws InterruptedException { + if (millis > 0L) { + Thread.sleep(millis); + } + } + + private void fail(final boolean isAppropriate, final Class<?> annotationClass) { + if (isAppropriate) { + throw new RuntimeException("Failure configured for " + annotationClass.getSimpleName() + " methods"); + } } + @Override public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { final ComponentLog logger = getLogger(); @@ -278,9 +337,7 @@ public class DebugFlow extends AbstractProcessor { // Make up to 2 passes to allow rollover from last cycle to first. // (This could be "while(true)" since responses should break out if selected, but this // prevents endless loops in the event of unexpected errors or future changes.) - int pass = 2; - while (pass > 0) { - pass -= 1; + for (int pass = 2; pass > 0; pass--) { if (ff == null) { if (curr_noff_resp.state() == NoFlowFileResponseState.NO_FF_SKIP_RESPONSE) { if (noFlowFileCurrSkip < noFlowFileMaxSkip) { @@ -438,18 +495,26 @@ public class DebugFlow extends AbstractProcessor { } } - private static Class<? extends RuntimeException> classNameToRuntimeExceptionClass(String name) { - Class<? extends RuntimeException> klass = null; - try { - Class<?> klass2 = Class.forName(name); - if (klass2 == RuntimeException.class || RuntimeException.class.isAssignableFrom(klass2)) { - //noinspection unchecked - klass = (Class<? extends RuntimeException>)klass2; + private static class RuntimeExceptionValidator implements Validator { + @Override + public ValidationResult validate(final String subject, final String input, final ValidationContext context) { + final ValidationResult.Builder resultBuilder = new ValidationResult.Builder() + .subject(subject) + .input(input); + + try { + final Class<?> exceptionClass = Class.forName(input); + if (RuntimeException.class.isAssignableFrom(exceptionClass)) { + resultBuilder.valid(true); + } else { + resultBuilder.valid(false).explanation("Class " + input + " is a Checked Exception, not a RuntimeException"); + } + } catch (ClassNotFoundException e) { + resultBuilder.valid(false).explanation("Class " + input + " cannot be found"); } - } catch (ClassNotFoundException e) { - klass = null; + + return resultBuilder.build(); } - return klass; } private enum FlowFileResponseState { http://git-wip-us.apache.org/repos/asf/nifi/blob/91a59a8a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDebugFlow.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDebugFlow.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDebugFlow.java index 5aa2e1e..c718942 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDebugFlow.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDebugFlow.java @@ -77,12 +77,7 @@ public class TestDebugFlow { } private boolean isInContents(byte[] content) { - for (Map.Entry entry : contents.entrySet()) { - if (((String)entry.getValue()).compareTo(new String(content)) == 0) { - return true; - } - } - return false; + return contents.containsValue(new String(content)); } @Test
