You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/03/11 18:55:32 UTC
[1/9] nifi git commit: NIFI-1464 fixed admin guide
Repository: nifi
Updated Branches:
refs/heads/master e1cf37fb8 -> a85c11953
NIFI-1464 fixed admin guide
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/8e7e2916
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/8e7e2916
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/8e7e2916
Branch: refs/heads/master
Commit: 8e7e2916fabcf0be86602d3f4c382718eed2b433
Parents: 59fac58
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Fri Mar 11 08:49:18 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
nifi-docs/src/main/asciidoc/administration-guide.adoc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e7e2916/nifi-docs/src/main/asciidoc/administration-guide.adoc
----------------------------------------------------------------------
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index 9ca582d..74e2e8a 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -1153,7 +1153,7 @@ nifi.nar.library.directory.lib2=/nars/lib2 +
Providing three total locations, including _nifi.nar.library.directory_.
|nifi.nar.working.directory|The location of the nar working directory. The default value is ./work/nar and probably should be left as is.
|nifi.documentation.working.directory|The documentation working directory. The default value is ./work/docs/components and probably should be left as is.
-|nifi.processor.scheduling.timeout|Time (milliseconds) to wait for a Processor's life-cycle operation (@OnScheduled and @OnUnscheduled) to finish before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
+|nifi.processor.scheduling.timeout|Time to wait for a Processor's life-cycle operation (@OnScheduled and @OnUnscheduled) to finish before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
|====
[8/9] nifi git commit: NIFI-1464 refactored the latest commit
Posted by ma...@apache.org.
NIFI-1464 refactored the latest commit
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/1c22f3f0
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/1c22f3f0
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/1c22f3f0
Branch: refs/heads/master
Commit: 1c22f3f0123632b544de484d9f42a02f64dfc768
Parents: 48af0bf
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Fri Mar 4 15:42:23 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
.../apache/nifi/controller/ProcessorNode.java | 22 ++++-----
.../nifi/controller/StandardFlowSerializer.java | 2 +-
.../scheduling/TestProcessorLifecycle.java | 48 ++++++++++----------
.../org/apache/nifi/web/api/dto/DtoFactory.java | 2 +-
4 files changed, 37 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/1c22f3f0/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 a428349..6c37418 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
@@ -113,17 +113,6 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
*/
@Override
public ScheduledState getScheduledState() {
- return this.scheduledState.get();
- }
-
- /**
- * Returns the logical state of this processor. Logical state ignores
- * transition states such as STOPPING and STARTING rounding it up to the
- * next logical state of STOPPED and RUNNING respectively.
- *
- * @return the logical state of this processor [DISABLED, STOPPED, RUNNING]
- */
- public ScheduledState getLogicalScheduledState() {
ScheduledState sc = this.scheduledState.get();
if (sc == ScheduledState.STARTING) {
return ScheduledState.RUNNING;
@@ -134,6 +123,17 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
}
/**
+ * Returns the physical state of this processor which includes transition
+ * states such as STOPPING and STARTING.
+ *
+ * @return the physical state of this processor [DISABLED, STOPPED, RUNNING,
+ * STARTIING, STOPIING]
+ */
+ public ScheduledState getPhysicalScheduledState() {
+ return this.scheduledState.get();
+ }
+
+ /**
* Will start the {@link Processor} represented by this
* {@link ProcessorNode}. Starting processor typically means invoking its
* operation that is annotated with @OnScheduled and then executing a
http://git-wip-us.apache.org/repos/asf/nifi/blob/1c22f3f0/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
index e9f958c..5d04cc2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
@@ -311,7 +311,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "yieldPeriod", processor.getYieldPeriod());
addTextElement(element, "bulletinLevel", processor.getBulletinLevel().toString());
addTextElement(element, "lossTolerant", String.valueOf(processor.isLossTolerant()));
- addTextElement(element, "scheduledState", processor.getLogicalScheduledState().name());
+ addTextElement(element, "scheduledState", processor.getScheduledState().name());
addTextElement(element, "schedulingStrategy", processor.getSchedulingStrategy().name());
addTextElement(element, "runDurationNanos", processor.getRunDuration(TimeUnit.NANOSECONDS));
http://git-wip-us.apache.org/repos/asf/nifi/blob/1c22f3f0/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
index d78009c..ca2e4fa 100644
--- 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
@@ -98,16 +98,16 @@ public class TestProcessorLifecycle {
UUID.randomUUID().toString());
assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
- assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getPhysicalScheduledState());
// validates idempotency
for (int i = 0; i < 2; i++) {
testProcNode.enable();
}
assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
- assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getPhysicalScheduledState());
testProcNode.disable();
assertEquals(ScheduledState.DISABLED, testProcNode.getScheduledState());
- assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.DISABLED, testProcNode.getPhysicalScheduledState());
fc.shutdown(true);
}
@@ -121,17 +121,17 @@ public class TestProcessorLifecycle {
UUID.randomUUID().toString());
testProcNode.setProperty("P", "hello");
assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
- assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getPhysicalScheduledState());
// validates idempotency
for (int i = 0; i < 2; i++) {
testProcNode.disable();
}
assertEquals(ScheduledState.DISABLED, testProcNode.getScheduledState());
- assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.DISABLED, testProcNode.getPhysicalScheduledState());
ProcessScheduler ps = fc.getProcessScheduler();
ps.startProcessor(testProcNode);
- assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+ assertEquals(ScheduledState.DISABLED, testProcNode.getPhysicalScheduledState());
fc.shutdown(true);
}
@@ -322,13 +322,13 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
- assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.RUNNING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
- assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.STOPPED);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
@@ -359,9 +359,9 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
@@ -392,12 +392,12 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STOPPING);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
@@ -422,12 +422,12 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STOPPING);
Thread.sleep(4000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
@@ -452,17 +452,17 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
ps.disableProcessor(testProcNode); // no effect
Thread.sleep(100);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
- assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.RUNNING);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
- assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
- assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.STOPPED);
+ assertTrue(testProcNode.getPhysicalScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
Thread.sleep(4000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
http://git-wip-us.apache.org/repos/asf/nifi/blob/1c22f3f0/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 64ffdaf..78584c3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -1644,7 +1644,7 @@ public final class DtoFactory {
dto.setType(node.getProcessor().getClass().getCanonicalName());
dto.setName(node.getName());
- dto.setState(node.getLogicalScheduledState().toString());
+ dto.setState(node.getScheduledState().toString());
// build the relationship dtos
final List<RelationshipDTO> relationships = new ArrayList<>();
[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
Posted by ma...@apache.org.
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
[9/9] nifi git commit: Merge branch 'master' of
https://git-wip-us.apache.org/repos/asf/nifi
Posted by ma...@apache.org.
Merge branch 'master' of https://git-wip-us.apache.org/repos/asf/nifi
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/a85c1195
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/a85c1195
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/a85c1195
Branch: refs/heads/master
Commit: a85c1195336a8c78c8e0d08d27dc429a881e3bc1
Parents: 8e7e291 e1cf37f
Author: Mark Payne <ma...@hotmail.com>
Authored: Fri Mar 11 12:55:23 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:55:23 2016 -0500
----------------------------------------------------------------------
.../elasticsearch/PutElasticsearch.java | 42 +++++++++++---------
.../elasticsearch/TestPutElasticsearch.java | 20 ++++++++++
2 files changed, 44 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
[7/9] nifi git commit: NIFI-1464 addressed latest PR comments
Posted by ma...@apache.org.
NIFI-1464 addressed latest PR comments
NIFI-1464 polishing
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/48af0bfb
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/48af0bfb
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/48af0bfb
Branch: refs/heads/master
Commit: 48af0bfbc5cafe06e003b2bd086b970cf49b5025
Parents: f53f45d
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Fri Feb 26 12:42:48 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
.../apache/nifi/controller/ProcessorNode.java | 55 ++++++++++++++++++
.../nifi/controller/StandardFlowSerializer.java | 4 +-
.../nifi/controller/StandardProcessorNode.java | 9 +--
.../scheduling/StandardProcessScheduler.java | 8 +--
.../scheduling/TestProcessorLifecycle.java | 61 +++++++++++++++++++-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 47 +++++++--------
6 files changed, 143 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/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 ff7977d..a428349 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
@@ -21,6 +21,7 @@ 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;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.connectable.Connectable;
@@ -31,12 +32,19 @@ import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class ProcessorNode extends AbstractConfiguredComponent implements Connectable {
+ private static final Logger logger = LoggerFactory.getLogger(ProcessorNode.class);
+
+ protected final AtomicReference<ScheduledState> scheduledState;
+
public ProcessorNode(final Processor processor, final String id,
final ValidationContextFactory validationContextFactory, final ControllerServiceProvider serviceProvider) {
super(processor, id, validationContextFactory, serviceProvider);
+ this.scheduledState = new AtomicReference<>(ScheduledState.STOPPED);
}
public abstract boolean isIsolated();
@@ -101,6 +109,31 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
public abstract void verifyCanStart(Set<ControllerServiceNode> ignoredReferences);
/**
+ *
+ */
+ @Override
+ public ScheduledState getScheduledState() {
+ return this.scheduledState.get();
+ }
+
+ /**
+ * Returns the logical state of this processor. Logical state ignores
+ * transition states such as STOPPING and STARTING rounding it up to the
+ * next logical state of STOPPED and RUNNING respectively.
+ *
+ * @return the logical state of this processor [DISABLED, STOPPED, RUNNING]
+ */
+ public ScheduledState getLogicalScheduledState() {
+ ScheduledState sc = this.scheduledState.get();
+ if (sc == ScheduledState.STARTING) {
+ return ScheduledState.RUNNING;
+ } else if (sc == ScheduledState.STOPPING) {
+ return ScheduledState.STOPPED;
+ }
+ return sc;
+ }
+
+ /**
* Will start the {@link Processor} represented by this
* {@link ProcessorNode}. Starting processor typically means invoking its
* operation that is annotated with @OnScheduled and then executing a
@@ -145,4 +178,26 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
*/
public abstract <T extends ProcessContext & ControllerServiceLookup> void stop(ScheduledExecutorService scheduler,
T processContext, Callable<Boolean> activeThreadMonitorCallback);
+
+ /**
+ * Will set the state of the processor to STOPPED which essentially implies
+ * that this processor can be started. This is idempotent operation and will
+ * result in the WARN message if processor can not be enabled.
+ */
+ public void enable() {
+ if (!this.scheduledState.compareAndSet(ScheduledState.DISABLED, ScheduledState.STOPPED)) {
+ logger.warn("Processor cannot be enabled because it is not disabled");
+ }
+ }
+
+ /**
+ * Will set the state of the processor to DISABLED which essentially implies
+ * that this processor can NOT be started. This is idempotent operation and
+ * will result in the WARN message if processor can not be enabled.
+ */
+ public void disable() {
+ if (!this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.DISABLED)) {
+ logger.warn("Processor cannot be disabled because its state is set to " + this.scheduledState);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
index 1ee85a2..e9f958c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowSerializer.java
@@ -255,7 +255,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "comments", port.getComments());
addTextElement(element, "scheduledState", port.getScheduledState().name());
addTextElement(element, "maxConcurrentTasks", port.getMaxConcurrentTasks());
- addTextElement(element, "useCompression", String.valueOf(((RemoteGroupPort) port).isUseCompression()));
+ addTextElement(element, "useCompression", String.valueOf(port.isUseCompression()));
parentElement.appendChild(element);
}
@@ -311,7 +311,7 @@ public class StandardFlowSerializer implements FlowSerializer {
addTextElement(element, "yieldPeriod", processor.getYieldPeriod());
addTextElement(element, "bulletinLevel", processor.getBulletinLevel().toString());
addTextElement(element, "lossTolerant", String.valueOf(processor.isLossTolerant()));
- addTextElement(element, "scheduledState", processor.getScheduledState().name());
+ addTextElement(element, "scheduledState", processor.getLogicalScheduledState().name());
addTextElement(element, "schedulingStrategy", processor.getSchedulingStrategy().name());
addTextElement(element, "runDurationNanos", processor.getRunDuration(TimeUnit.NANOSECONDS));
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/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 dac56b5..2a0819c 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
@@ -106,7 +106,6 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
private final AtomicReference<List<Connection>> incomingConnectionsRef;
private final AtomicBoolean isolated;
private final AtomicBoolean lossTolerant;
- private final AtomicReference<ScheduledState> scheduledState;
private final AtomicReference<String> comments;
private final AtomicReference<Position> position;
private final AtomicReference<String> annotationData;
@@ -145,7 +144,6 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
destinations = new HashMap<>();
connections = new HashMap<>();
incomingConnectionsRef = new AtomicReference<List<Connection>>(new ArrayList<Connection>());
- scheduledState = new AtomicReference<>(ScheduledState.STOPPED);
lossTolerant = new AtomicBoolean(false);
final Set<Relationship> emptySetOfRelationships = new HashSet<>();
undefinedRelationshipsToTerminate = new AtomicReference<>(emptySetOfRelationships);
@@ -214,11 +212,6 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
@Override
- public ScheduledState getScheduledState() {
- return this.scheduledState.get();
- }
-
- @Override
public Position getPosition() {
return position.get();
}
@@ -1276,7 +1269,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
};
taskScheduler.execute(startProcRunnable);
} else {
- LOG.warn("Can not start Processor since it's already in the process of being started");
+ LOG.warn("Can not start Processor since it's already in the process of being started or it is DISABLED");
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/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 84d667f..bbaa23b 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
@@ -473,16 +473,12 @@ public final class StandardProcessScheduler implements ProcessScheduler {
@Override
public synchronized void enableProcessor(final ProcessorNode procNode) {
- if (procNode.getScheduledState() != ScheduledState.DISABLED) {
- throw new IllegalStateException("Processor cannot be enabled because it is not disabled");
- }
+ procNode.enable();
}
@Override
public synchronized void disableProcessor(final ProcessorNode procNode) {
- if (procNode.getScheduledState() != ScheduledState.STOPPED) {
- throw new IllegalStateException("Processor cannot be disabled because its state is set to " + procNode.getScheduledState());
- }
+ procNode.disable();
}
public synchronized void enableReportingTask(final ReportingTaskNode taskNode) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/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
index 2cce14d..d78009c 100644
--- 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
@@ -89,6 +89,53 @@ public class TestProcessorLifecycle {
FileUtils.deleteDirectory(new File("./target/content_repository"));
}
+ @Test
+ public void validateEnableOperation() 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());
+
+ assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ // validates idempotency
+ for (int i = 0; i < 2; i++) {
+ testProcNode.enable();
+ }
+ assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ testProcNode.disable();
+ assertEquals(ScheduledState.DISABLED, testProcNode.getScheduledState());
+ assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+ fc.shutdown(true);
+ }
+
+
+ @Test
+ public void validateDisableOperation() 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");
+ assertEquals(ScheduledState.STOPPED, testProcNode.getScheduledState());
+ assertEquals(ScheduledState.STOPPED, testProcNode.getLogicalScheduledState());
+ // validates idempotency
+ for (int i = 0; i < 2; i++) {
+ testProcNode.disable();
+ }
+ assertEquals(ScheduledState.DISABLED, testProcNode.getScheduledState());
+ assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+
+ ProcessScheduler ps = fc.getProcessScheduler();
+ ps.startProcessor(testProcNode);
+ assertEquals(ScheduledState.DISABLED, testProcNode.getLogicalScheduledState());
+
+ fc.shutdown(true);
+ }
+
/**
* Will validate the idempotent nature of processor start operation which
* can be called multiple times without any side-effects.
@@ -276,10 +323,12 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.STOPPED);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
@@ -360,7 +409,7 @@ public class TestProcessorLifecycle {
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyInterruptable() throws Exception {
- NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5000");
+ NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5 sec");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
@@ -390,7 +439,7 @@ public class TestProcessorLifecycle {
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyUninterruptable() throws Exception {
- NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5000");
+ NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5 sec");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
@@ -406,9 +455,14 @@ public class TestProcessorLifecycle {
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ ps.disableProcessor(testProcNode); // no effect
+ Thread.sleep(100);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.STARTING);
+ assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(100);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPING);
+ assertTrue(testProcNode.getLogicalScheduledState() == ScheduledState.STOPPED);
Thread.sleep(4000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
fc.shutdown(true);
@@ -435,6 +489,9 @@ public class TestProcessorLifecycle {
ps.startProcessor(testProcNode);
Thread.sleep(1000);
assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
+ ps.disableProcessor(testProcNode);
+ Thread.sleep(100);
+ assertTrue(testProcNode.getScheduledState() == ScheduledState.RUNNING);
ps.stopProcessor(testProcNode);
Thread.sleep(500);
assertTrue(testProcNode.getScheduledState() == ScheduledState.STOPPED);
http://git-wip-us.apache.org/repos/asf/nifi/blob/48af0bfb/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 652fd04..64ffdaf 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -16,6 +16,29 @@
*/
package org.apache.nifi.web.api.dto;
+import java.text.Collator;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.WebApplicationException;
+
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.action.Action;
import org.apache.nifi.action.component.details.ComponentDetails;
@@ -128,28 +151,6 @@ import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO;
import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.StatusDTO;
-import javax.ws.rs.WebApplicationException;
-import java.text.Collator;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.TreeSet;
-import java.util.concurrent.TimeUnit;
-
public final class DtoFactory {
@SuppressWarnings("rawtypes")
@@ -1643,7 +1644,7 @@ public final class DtoFactory {
dto.setType(node.getProcessor().getClass().getCanonicalName());
dto.setName(node.getName());
- dto.setState(node.getScheduledState().toString());
+ dto.setState(node.getLogicalScheduledState().toString());
// build the relationship dtos
final List<RelationshipDTO> relationships = new ArrayList<>();
[2/9] nifi git commit: NIFI-1464 ensured that OnUnscheduled is
treated the same as OnScheduled
Posted by ma...@apache.org.
NIFI-1464 ensured that OnUnscheduled is treated the same as OnScheduled
NIFI-1464 polished javadocs, error messages and docs
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/59fac58c
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/59fac58c
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/59fac58c
Branch: refs/heads/master
Commit: 59fac58c96f4127b1bf6fb7a501e934692534a03
Parents: 1c22f3f
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Wed Mar 9 12:30:11 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
.../org/apache/nifi/util/NiFiProperties.java | 2 +-
.../src/main/asciidoc/administration-guide.adoc | 2 +-
.../nifi/controller/StandardProcessorNode.java | 79 +++++++++++---------
.../scheduling/TestProcessorLifecycle.java | 4 +-
4 files changed, 49 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/59fac58c/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 2be8177..3d05a47 100644
--- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -71,7 +71,7 @@ public class NiFiProperties extends Properties {
public static final String ADMINISTRATIVE_YIELD_DURATION = "nifi.administrative.yield.duration";
public static final String PERSISTENT_STATE_DIRECTORY = "nifi.persistent.state.directory";
public static final String BORED_YIELD_DURATION = "nifi.bored.yield.duration";
- public static final String PROCESSOR_START_TIMEOUT = "nifi.processor.start.timeout";
+ public static final String PROCESSOR_SCHEDULING_TIMEOUT = "nifi.processor.scheduling.timeout";
// content repository properties
public static final String REPOSITORY_CONTENT_PREFIX = "nifi.content.repository.directory.";
http://git-wip-us.apache.org/repos/asf/nifi/blob/59fac58c/nifi-docs/src/main/asciidoc/administration-guide.adoc
----------------------------------------------------------------------
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index 58fe205..9ca582d 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -1153,7 +1153,7 @@ nifi.nar.library.directory.lib2=/nars/lib2 +
Providing three total locations, including _nifi.nar.library.directory_.
|nifi.nar.working.directory|The location of the nar working directory. The default value is ./work/nar and probably should be left as is.
|nifi.documentation.working.directory|The documentation working directory. The default value is ./work/docs/components and probably should be left as is.
-|nifi.processor.start.timeout|Time (milliseconds) to wait for a Processors to start before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
+|nifi.processor.scheduling.timeout|Time (milliseconds) to wait for a Processor's life-cycle operation (@OnScheduled and @OnUnscheduled) to finish before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
|====
http://git-wip-us.apache.org/repos/asf/nifi/blob/59fac58c/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 2a0819c..85c5fe8 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
@@ -1241,9 +1241,16 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void run() {
try {
- SchedulingContext schedulingContext = new StandardSchedulingContext(processContext, getControllerServiceProvider(),
+ final SchedulingContext schedulingContext = new StandardSchedulingContext(processContext, getControllerServiceProvider(),
StandardProcessorNode.this, processContext.getStateManager());
- invokeOnScheduleAsync(taskScheduler, schedulingContext);
+ invokeTaskAsCancelableFuture(taskScheduler, new Callable<Void>() {
+ @SuppressWarnings("deprecation")
+ @Override
+ public Void call() throws Exception {
+ ReflectionUtils.invokeMethodsWithAnnotations(OnScheduled.class, org.apache.nifi.processor.annotation.OnScheduled.class, processor, schedulingContext);
+ return null;
+ }
+ });
if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) {
schedulingAgentCallback.run(); // callback provided by StandardProcessScheduler to essentially initiate component's onTrigger() cycle
} else { // can only happen if stopProcessor was called before service was transitioned to RUNNING state
@@ -1302,12 +1309,20 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
public <T extends ProcessContext & ControllerServiceLookup> void stop(final ScheduledExecutorService scheduler,
final T processContext, final Callable<Boolean> activeThreadMonitorCallback) {
if (this.scheduledState.compareAndSet(ScheduledState.RUNNING, ScheduledState.STOPPING)) { // will ensure that the Processor represented by this node can only be stopped once
- final Runnable stopProcRunnable = new Runnable() {
+ invokeTaskAsCancelableFuture(scheduler, new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
+ return null;
+ }
+ });
+ // will continue to monitor active threads, invoking OnStopped once
+ // there are none
+ scheduler.execute(new Runnable() {
@Override
public void run() {
try {
if (activeThreadMonitorCallback.call()) {
- ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext);
scheduledState.set(ScheduledState.STOPPED);
} else {
@@ -1317,8 +1332,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
LOG.warn("Failed while shutting down processor " + processor, e);
}
}
- };
- scheduler.execute(stopProcRunnable);
+ });
} else {
/*
* We do compareAndSet() instead of set() to ensure that Processor
@@ -1333,41 +1347,35 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * Will invoke processor's methods annotated with @OnSchedule asynchronously
- * to ensure that it could be interrupted if stop action was initiated on
- * the processor that may be sitting in the infinitely blocking @OnSchedule
+ * Will invoke lifecycle operation (OnScheduled or OnUnscheduled)
+ * asynchronously to ensure that it could be interrupted if stop action was
+ * initiated on the processor that may be infinitely blocking in such
* operation. While this approach paves the way for further enhancements
* related to managing processor'slife-cycle operation at the moment the
* interrupt will not happen automatically. This is primarily to preserve
- * the existing behavior or the NiFi where stop operation can only be
+ * the existing behavior of the NiFi where stop operation can only be
* invoked once the processor is started. Unfortunately that could mean that
- * the processor may be blocking indefinitely in the @Oncheduled call. To
- * deal with that a new NiFi property has been introduced
- * <i>nifi.processor.start.timeout</i> which allows one to set the time (in
- * milliseconds) of how long to wait before canceling the @OnScheduled task
- * allowing processor's stop sequence to proceed. The default value for this
- * property is {@link Long#MAX_VALUE}.
+ * the processor may be blocking indefinitely in lifecycle operation
+ * (OnScheduled or OnUnscheduled). To deal with that a new NiFi property has
+ * been introduced <i>nifi.processor.scheduling.timeout</i> which allows one
+ * to set the time (in milliseconds) of how long to wait before canceling
+ * such lifecycle operation (OnScheduled or OnUnscheduled) allowing
+ * processor's stop sequence to proceed. The default value for this property
+ * is {@link Long#MAX_VALUE}.
* <p>
* NOTE: Canceling the task does not guarantee that the task will actually
* completes (successfully or otherwise), since cancellation of the task
- * will issue a simple Thread.interrupt(). However code inside
- * of @OnScheduled operation is written purely and will ignore thread
- * interrupts you may end up with runaway thread which may eventually
- * require NiFi reboot. In any event, the above explanation will be logged
- * (WARN) informing a user so further actions could be taken.
+ * will issue a simple Thread.interrupt(). However code inside of lifecycle
+ * operation (OnScheduled or OnUnscheduled) is written purely and will
+ * ignore thread interrupts you may end up with runaway thread which may
+ * eventually require NiFi reboot. In any event, the above explanation will
+ * be logged (WARN) informing a user so further actions could be taken.
* </p>
*/
- private void invokeOnScheduleAsync(ScheduledExecutorService taskScheduler, final SchedulingContext schedulingContext) throws ExecutionException {
- Future<Void> executionResult = taskScheduler.submit(new Callable<Void>() {
- @SuppressWarnings("deprecation")
- @Override
- public Void call() throws Exception {
- ReflectionUtils.invokeMethodsWithAnnotations(OnScheduled.class, org.apache.nifi.processor.annotation.OnScheduled.class, processor, schedulingContext);
- return null;
- }
- });
+ private void invokeTaskAsCancelableFuture(ScheduledExecutorService taskScheduler, Callable<Void> task) {
+ Future<Void> executionResult = taskScheduler.submit(task);
- String timeoutString = NiFiProperties.getInstance().getProperty(NiFiProperties.PROCESSOR_START_TIMEOUT);
+ String timeoutString = NiFiProperties.getInstance().getProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT);
long onScheduleTimeout = timeoutString == null ? Long.MAX_VALUE
: FormatUtils.getTimeDuration(timeoutString.trim(), TimeUnit.MILLISECONDS);
@@ -1375,17 +1383,20 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
executionResult.get(onScheduleTimeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
LOG.warn("Thread was interrupted while waiting for processor '" + this.processor.getClass().getSimpleName()
- + "' @OnSchedule operation to finish.");
+ + "' lifecycle operation (OnScheduled or OnUnscheduled) to finish.");
Thread.currentThread().interrupt();
} catch (TimeoutException e) {
executionResult.cancel(true);
- LOG.warn("Timed out while waiting for the task executing @OnSchedule operation for '"
+ LOG.warn("Timed out while waiting for lifecycle operation (OnScheduled or OnUnscheduled) of '"
+ 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 @OnSchedule method may "
+ + "guarantee that the task will be canceled since the code inside current lifecycle operation (OnScheduled or OnUnscheduled) may "
+ "have been written to ignore interrupts which may result in runaway thread which 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.");
+ } catch (ExecutionException e){
+ throw new RuntimeException(
+ "Failed while executing one of processor's lifecycle tasks (OnScheduled or OnUnscheduled).", e);
}
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/59fac58c/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
index ca2e4fa..560c4cb 100644
--- 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
@@ -409,7 +409,7 @@ public class TestProcessorLifecycle {
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyInterruptable() throws Exception {
- NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5 sec");
+ NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
@@ -439,7 +439,7 @@ public class TestProcessorLifecycle {
*/
@Test
public void validateProcessorCanBeStoppedWhenOnScheduledBlocksIndefinitelyUninterruptable() throws Exception {
- NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, "5 sec");
+ NiFiProperties.getInstance().setProperty(NiFiProperties.PROCESSOR_SCHEDULING_TIMEOUT, "5 sec");
FlowController fc = this.buildFlowControllerForTest();
ProcessGroup testGroup = fc.createProcessGroup(UUID.randomUUID().toString());
this.setControllerRootGroup(fc, testGroup);
[3/9] nifi git commit: NIFI-1464 addressed PR comments from @apiri
and @markap14
Posted by ma...@apache.org.
NIFI-1464 addressed PR comments from @apiri and @markap14
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/f53f45de
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/f53f45de
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/f53f45de
Branch: refs/heads/master
Commit: f53f45def3ed353232ec65d656bba4eee922b252
Parents: 0c5b1c2
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Wed Feb 17 09:10:15 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
.../apache/nifi/controller/ProcessorNode.java | 4 +-
.../nifi/controller/StandardProcessorNode.java | 6 +-
.../scheduling/AbstractSchedulingAgent.java | 2 +-
.../StandardControllerServiceProvider.java | 1 -
.../TestStandardControllerServiceProvider.java | 122 -------------------
5 files changed, 7 insertions(+), 128 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/f53f45de/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 d7dbe24..ff7977d 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
@@ -102,7 +102,7 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
/**
* Will start the {@link Processor} represented by this
- * {@link ProcessorNode}. Starting processor typically means invoking it's
+ * {@link ProcessorNode}. Starting processor typically means invoking its
* operation that is annotated with @OnScheduled and then executing a
* callback provided by the {@link ProcessScheduler} to which typically
* initiates
@@ -126,7 +126,7 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
/**
* Will stop the {@link Processor} represented by this {@link ProcessorNode}.
- * Stopping processor typically means invoking it's operation that is
+ * Stopping processor typically means invoking its operation that is
* annotated with @OnUnschedule and then @OnStopped.
*
* @param scheduler
http://git-wip-us.apache.org/repos/asf/nifi/blob/f53f45de/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 3e07995..dac56b5 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
@@ -1374,8 +1374,10 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
});
- long onScheduleTimeout = Long.parseLong(NiFiProperties.getInstance()
- .getProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, String.valueOf(Long.MAX_VALUE)));
+ String timeoutString = NiFiProperties.getInstance().getProperty(NiFiProperties.PROCESSOR_START_TIMEOUT);
+ long onScheduleTimeout = timeoutString == null ? Long.MAX_VALUE
+ : FormatUtils.getTimeDuration(timeoutString.trim(), TimeUnit.MILLISECONDS);
+
try {
executionResult.get(onScheduleTimeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/f53f45de/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
index b931c64..3544dac 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
@@ -23,7 +23,7 @@ import org.apache.nifi.controller.ReportingTaskNode;
* Base implementation of the {@link SchedulingAgent} which encapsulates the
* updates to the {@link ScheduleState} based on invoked operation and then
* delegates to the corresponding 'do' methods. For example; By invoking
- * {@link #schedule(Connectable, ScheduleState)} the the
+ * {@link #schedule(Connectable, ScheduleState)} the
* {@link ScheduleState#setScheduled(boolean)} with value 'true' will be
* invoked.
*
http://git-wip-us.apache.org/repos/asf/nifi/blob/f53f45de/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 3b9b073..77dc87e 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,7 +457,6 @@ 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/f53f45de/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 9b35238..11b73a8 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
@@ -17,7 +17,6 @@
package org.apache.nifi.controller.service;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import java.beans.PropertyDescriptor;
@@ -41,15 +40,11 @@ import org.apache.nifi.controller.service.mock.ServiceA;
import org.apache.nifi.controller.service.mock.ServiceB;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.StandardProcessGroup;
-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;
-import org.mockito.stubbing.Answer;
public class TestStandardControllerServiceProvider {
private static StateManagerProvider stateManagerProvider = new StateManagerProvider() {
@@ -191,119 +186,6 @@ public class TestStandardControllerServiceProvider {
}
}
- @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);
-
- // build a graph of reporting tasks and controller services with dependencies as such:
- //
- // Processor P1 -> A -> B -> D
- // Processor P2 -> C ---^----^
- //
- // In other words, Processor P1 references Controller Service A, which references B, which references D.
- // AND
- // Processor P2 references Controller Service C, which references B and D.
- //
- // So we have to verify that if D is enabled, when we enable its referencing services,
- // we enable C and B, even if we attempt to enable C before B... i.e., if we try to enable C, we cannot do so
- // until B is first enabled so ensure that we enable B first.
- final ControllerServiceNode serviceNode1 = provider.createControllerService(ServiceA.class.getName(), "1", false);
- final ControllerServiceNode serviceNode2 = provider.createControllerService(ServiceA.class.getName(), "2", false);
- final ControllerServiceNode serviceNode3 = provider.createControllerService(ServiceA.class.getName(), "3", false);
- final ControllerServiceNode serviceNode4 = provider.createControllerService(ServiceB.class.getName(), "4", false);
-
- final ProcessGroup mockProcessGroup = Mockito.mock(ProcessGroup.class);
- Mockito.doAnswer(new Answer<Object>() {
- @Override
- public Object answer(InvocationOnMock invocation) throws Throwable {
- final ProcessorNode procNode = (ProcessorNode) invocation.getArguments()[0];
- procNode.verifyCanStart();
- // procNode.setScheduledState(ScheduledState.RUNNING);
- return null;
- }
- }).when(mockProcessGroup).startProcessor(Mockito.any(ProcessorNode.class));
-
- Mockito.doAnswer(new Answer<Object>() {
- @Override
- public Object answer(final InvocationOnMock invocation) throws Throwable {
- final ProcessorNode procNode = (ProcessorNode) invocation.getArguments()[0];
- procNode.verifyCanStop();
- // procNode.setScheduledState(ScheduledState.STOPPED);
- return null;
- }
- }).when(mockProcessGroup).stopProcessor(Mockito.any(ProcessorNode.class));
-
- final String id1 = UUID.randomUUID().toString();
- final ProcessorNode procNodeA = new StandardProcessorNode(new DummyProcessor(), id1,
- new StandardValidationContextFactory(provider), scheduler, provider);
- procNodeA.getProcessor().initialize(new StandardProcessorInitializationContext(id1, null, provider));
- procNodeA.setProperty(DummyProcessor.SERVICE.getName(), "1");
- procNodeA.setProcessGroup(mockProcessGroup);
-
- final String id2 = UUID.randomUUID().toString();
- final ProcessorNode procNodeB = new StandardProcessorNode(new DummyProcessor(), id2,
- new StandardValidationContextFactory(provider), scheduler, provider);
- procNodeB.getProcessor().initialize(new StandardProcessorInitializationContext(id2, null, provider));
- procNodeB.setProperty(DummyProcessor.SERVICE.getName(), "3");
- procNodeB.setProcessGroup(mockProcessGroup);
-
- serviceNode1.setProperty(ServiceA.OTHER_SERVICE.getName(), "2");
- serviceNode2.setProperty(ServiceA.OTHER_SERVICE.getName(), "4");
- serviceNode3.setProperty(ServiceA.OTHER_SERVICE.getName(), "2");
- serviceNode3.setProperty(ServiceA.OTHER_SERVICE_2.getName(), "4");
-
- provider.enableControllerService(serviceNode4);
- provider.enableReferencingServices(serviceNode4);
- provider.scheduleReferencingComponents(serviceNode4);
-
- final Set<ControllerServiceState> enableStates = new HashSet<>();
- enableStates.add(ControllerServiceState.ENABLED);
- enableStates.add(ControllerServiceState.ENABLING);
-
- while (serviceNode3.getState() != ControllerServiceState.ENABLED
- || serviceNode2.getState() != ControllerServiceState.ENABLED
- || serviceNode1.getState() != ControllerServiceState.ENABLED) {
- assertTrue(enableStates.contains(serviceNode3.getState()));
- assertTrue(enableStates.contains(serviceNode2.getState()));
- assertTrue(enableStates.contains(serviceNode1.getState()));
- }
- assertTrue(procNodeA.isRunning());
- assertTrue(procNodeB.isRunning());
-
- // stop processors and verify results.
- provider.unscheduleReferencingComponents(serviceNode4);
- assertFalse(procNodeA.isRunning());
- assertFalse(procNodeB.isRunning());
- while (serviceNode3.getState() != ControllerServiceState.ENABLED
- || serviceNode2.getState() != ControllerServiceState.ENABLED
- || serviceNode1.getState() != ControllerServiceState.ENABLED) {
- assertTrue(enableStates.contains(serviceNode3.getState()));
- assertTrue(enableStates.contains(serviceNode2.getState()));
- assertTrue(enableStates.contains(serviceNode1.getState()));
- }
-
- provider.disableReferencingServices(serviceNode4);
- final Set<ControllerServiceState> disableStates = new HashSet<>();
- disableStates.add(ControllerServiceState.DISABLED);
- disableStates.add(ControllerServiceState.DISABLING);
-
- // Wait for the services to be disabled.
- while (serviceNode3.getState() != ControllerServiceState.DISABLED
- || serviceNode2.getState() != ControllerServiceState.DISABLED
- || serviceNode1.getState() != ControllerServiceState.DISABLED) {
- assertTrue(disableStates.contains(serviceNode3.getState()));
- assertTrue(disableStates.contains(serviceNode2.getState()));
- assertTrue(disableStates.contains(serviceNode1.getState()));
- }
-
- assertEquals(ControllerServiceState.ENABLED, serviceNode4.getState());
-
- provider.disableControllerService(serviceNode4);
- assertTrue(disableStates.contains(serviceNode4.getState()));
- }
@Test
public void testOrderingOfServices() {
@@ -476,9 +358,5 @@ public class TestStandardControllerServiceProvider {
// procNode.setScheduledState(ScheduledState.RUNNING);
provider.unscheduleReferencingComponents(serviceNode);
assertEquals(ScheduledState.STOPPED, procNode.getScheduledState());
-
- // procNode.setScheduledState(ScheduledState.DISABLED);
- // provider.unscheduleReferencingComponents(serviceNode);
- // assertEquals(ScheduledState.DISABLED, procNode.getScheduledState());
}
}
[5/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
Posted by ma...@apache.org.
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/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 f57343e..3e07995 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
@@ -18,6 +18,7 @@ package org.apache.nifi.controller;
import static java.util.Objects.requireNonNull;
+import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -27,13 +28,16 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder;
@@ -46,6 +50,9 @@ import org.apache.nifi.annotation.behavior.TriggerSerially;
import org.apache.nifi.annotation.behavior.TriggerWhenAnyDestinationAvailable;
import org.apache.nifi.annotation.behavior.TriggerWhenEmpty;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
+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.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.connectable.Connectable;
@@ -57,23 +64,34 @@ import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.logging.LogRepositoryFactory;
+import org.apache.nifi.logging.ProcessorLog;
import org.apache.nifi.nar.NarCloseable;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.SchedulingContext;
+import org.apache.nifi.processor.SimpleProcessLogger;
+import org.apache.nifi.processor.StandardSchedulingContext;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.FormatUtils;
+import org.apache.nifi.util.NiFiProperties;
+import org.apache.nifi.util.ReflectionUtils;
import org.quartz.CronExpression;
+import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * ProcessorNode provides thread-safe access to a FlowFileProcessor as it exists within a controlled flow. This node keeps track of the processor, its scheduling information and its relationships to
- * other processors and whatever scheduled futures exist for it. Must be thread safe.
+ * ProcessorNode provides thread-safe access to a FlowFileProcessor as it exists
+ * within a controlled flow. This node keeps track of the processor, its
+ * scheduling information and its relationships to other processors and whatever
+ * scheduled futures exist for it. Must be thread safe.
*
*/
public class StandardProcessorNode extends ProcessorNode implements Connectable {
+ private static final Logger LOG = LoggerFactory.getLogger(StandardProcessorNode.class);
+
public static final String BULLETIN_OBSERVER_ID = "bulletin-observer";
public static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.MILLISECONDS;
@@ -86,16 +104,16 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
private final Map<Relationship, Set<Connection>> connections;
private final AtomicReference<Set<Relationship>> undefinedRelationshipsToTerminate;
private final AtomicReference<List<Connection>> incomingConnectionsRef;
- private final ReentrantReadWriteLock rwLock;
- private final Lock readLock;
- private final Lock writeLock;
private final AtomicBoolean isolated;
private final AtomicBoolean lossTolerant;
private final AtomicReference<ScheduledState> scheduledState;
private final AtomicReference<String> comments;
private final AtomicReference<Position> position;
private final AtomicReference<String> annotationData;
- private final AtomicReference<String> schedulingPeriod; // stored as string so it's presented to user as they entered it
+ private final AtomicReference<String> schedulingPeriod; // stored as string
+ // so it's presented
+ // to user as they
+ // entered it
private final AtomicReference<String> yieldPeriod;
private final AtomicReference<String> penalizationPeriod;
private final AtomicReference<Map<String, String>> style;
@@ -114,10 +132,12 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
private long runNanos = 0L;
private SchedulingStrategy schedulingStrategy; // guarded by read/write lock
+ // ??????? NOT any more
@SuppressWarnings("deprecation")
- public StandardProcessorNode(final Processor processor, final String uuid, final ValidationContextFactory validationContextFactory,
- final ProcessScheduler scheduler, final ControllerServiceProvider controllerServiceProvider) {
+ public StandardProcessorNode(final Processor processor, final String uuid,
+ final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
+ final ControllerServiceProvider controllerServiceProvider) {
super(processor, uuid, validationContextFactory, controllerServiceProvider);
this.processor = processor;
@@ -126,9 +146,6 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
connections = new HashMap<>();
incomingConnectionsRef = new AtomicReference<List<Connection>>(new ArrayList<Connection>());
scheduledState = new AtomicReference<>(ScheduledState.STOPPED);
- rwLock = new ReentrantReadWriteLock(false);
- readLock = rwLock.readLock();
- writeLock = rwLock.writeLock();
lossTolerant = new AtomicBoolean(false);
final Set<Relationship> emptySetOfRelationships = new HashSet<>();
undefinedRelationshipsToTerminate = new AtomicReference<>(emptySetOfRelationships);
@@ -147,15 +164,21 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
penalizationPeriod = new AtomicReference<>(DEFAULT_PENALIZATION_PERIOD);
final Class<?> procClass = processor.getClass();
- triggerWhenEmpty = procClass.isAnnotationPresent(TriggerWhenEmpty.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenEmpty.class);
- sideEffectFree = procClass.isAnnotationPresent(SideEffectFree.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SideEffectFree.class);
- batchSupported = procClass.isAnnotationPresent(SupportsBatching.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SupportsBatching.class);
- triggeredSerially = procClass.isAnnotationPresent(TriggerSerially.class) || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerSerially.class);
+ triggerWhenEmpty = procClass.isAnnotationPresent(TriggerWhenEmpty.class)
+ || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenEmpty.class);
+ sideEffectFree = procClass.isAnnotationPresent(SideEffectFree.class)
+ || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SideEffectFree.class);
+ batchSupported = procClass.isAnnotationPresent(SupportsBatching.class)
+ || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.SupportsBatching.class);
+ triggeredSerially = procClass.isAnnotationPresent(TriggerSerially.class)
+ || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerSerially.class);
triggerWhenAnyDestinationAvailable = procClass.isAnnotationPresent(TriggerWhenAnyDestinationAvailable.class)
- || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.TriggerWhenAnyDestinationAvailable.class);
+ || procClass.isAnnotationPresent(
+ org.apache.nifi.processor.annotation.TriggerWhenAnyDestinationAvailable.class);
this.validationContextFactory = validationContextFactory;
eventDrivenSupported = (procClass.isAnnotationPresent(EventDriven.class)
- || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.EventDriven.class)) && !triggeredSerially && !triggerWhenEmpty;
+ || procClass.isAnnotationPresent(org.apache.nifi.processor.annotation.EventDriven.class))
+ && !triggeredSerially && !triggerWhenEmpty;
final boolean inputRequirementPresent = procClass.isAnnotationPresent(InputRequirement.class);
if (inputRequirementPresent) {
@@ -176,26 +199,23 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * Provides and opportunity to retain information about this particular processor instance
+ * Provides and opportunity to retain information about this particular
+ * processor instance
*
- * @param comments new comments
+ * @param comments
+ * new comments
*/
@Override
public void setComments(final String comments) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- this.comments.set(comments);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ this.comments.set(comments);
}
@Override
public ScheduledState getScheduledState() {
- return scheduledState.get();
+ return this.scheduledState.get();
}
@Override
@@ -226,7 +246,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return if true flow file content generated by this processor is considered loss tolerant
+ * @return if true flow file content generated by this processor is
+ * considered loss tolerant
*/
@Override
public boolean isLossTolerant() {
@@ -239,7 +260,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return true if the processor has the {@link TriggerWhenEmpty} annotation, false otherwise.
+ * @return true if the processor has the {@link TriggerWhenEmpty}
+ * annotation, false otherwise.
*/
@Override
public boolean isTriggerWhenEmpty() {
@@ -247,7 +269,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return true if the processor has the {@link SideEffectFree} annotation, false otherwise.
+ * @return true if the processor has the {@link SideEffectFree} annotation,
+ * false otherwise.
*/
@Override
public boolean isSideEffectFree() {
@@ -260,7 +283,9 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return true if the processor has the {@link TriggerWhenAnyDestinationAvailable} annotation, false otherwise.
+ * @return true if the processor has the
+ * {@link TriggerWhenAnyDestinationAvailable} annotation, false
+ * otherwise.
*/
@Override
public boolean isTriggerWhenAnyDestinationAvailable() {
@@ -268,38 +293,31 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * Indicates whether flow file content made by this processor must be persisted
+ * Indicates whether flow file content made by this processor must be
+ * persisted
*
- * @param lossTolerant tolerant
+ * @param lossTolerant
+ * tolerant
*/
@Override
public void setLossTolerant(final boolean lossTolerant) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- this.lossTolerant.set(lossTolerant);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ this.lossTolerant.set(lossTolerant);
}
/**
* Indicates whether the processor runs on only the primary node.
*
- * @param isolated isolated
+ * @param isolated
+ * isolated
*/
public void setIsolated(final boolean isolated) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- this.isolated.set(isolated);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ this.isolated.set(isolated);
}
@Override
@@ -308,33 +326,28 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
return true;
}
final Set<Relationship> terminatable = undefinedRelationshipsToTerminate.get();
- if (terminatable == null) {
- return false;
- }
- return terminatable.contains(relationship);
+ return terminatable == null ? false : terminatable.contains(relationship);
}
@Override
public void setAutoTerminatedRelationships(final Set<Relationship> terminate) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
+ }
- for (final Relationship rel : terminate) {
- if (!getConnections(rel).isEmpty()) {
- throw new IllegalStateException("Cannot mark relationship '" + rel.getName() + "' as auto-terminated because Connection already exists with this relationship");
- }
+ for (final Relationship rel : terminate) {
+ if (!getConnections(rel).isEmpty()) {
+ throw new IllegalStateException("Cannot mark relationship '" + rel.getName()
+ + "' as auto-terminated because Connection already exists with this relationship");
}
- undefinedRelationshipsToTerminate.set(new HashSet<>(terminate));
- } finally {
- writeLock.unlock();
}
+ undefinedRelationshipsToTerminate.set(new HashSet<>(terminate));
}
/**
- * @return an unmodifiable Set that contains all of the ProcessorRelationship objects that are configured to be auto-terminated
+ * @return an unmodifiable Set that contains all of the
+ * ProcessorRelationship objects that are configured to be
+ * auto-terminated
*/
@Override
public Set<Relationship> getAutoTerminatedRelationships() {
@@ -346,7 +359,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return the value of the processor's {@link CapabilityDescription} annotation, if one exists, else <code>null</code>.
+ * @return the value of the processor's {@link CapabilityDescription}
+ * annotation, if one exists, else <code>null</code>.
*/
@SuppressWarnings("deprecation")
public String getProcessorDescription() {
@@ -355,7 +369,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
if (capDesc != null) {
description = capDesc.value();
} else {
- final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapDesc = processor.getClass().getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class);
+ final org.apache.nifi.processor.annotation.CapabilityDescription deprecatedCapDesc = processor.getClass()
+ .getAnnotation(org.apache.nifi.processor.annotation.CapabilityDescription.class);
if (deprecatedCapDesc != null) {
description = deprecatedCapDesc.value();
}
@@ -366,20 +381,19 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void setName(final String name) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- super.setName(name);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ super.setName(name);
}
/**
- * @param timeUnit determines the unit of time to represent the scheduling period. If null will be reported in units of {@link #DEFAULT_SCHEDULING_TIME_UNIT}
- * @return the schedule period that should elapse before subsequent cycles of this processor's tasks
+ * @param timeUnit
+ * determines the unit of time to represent the scheduling
+ * period. If null will be reported in units of
+ * {@link #DEFAULT_SCHEDULING_TIME_UNIT}
+ * @return the schedule period that should elapse before subsequent cycles
+ * of this processor's tasks
*/
@Override
public long getSchedulingPeriod(final TimeUnit timeUnit) {
@@ -388,37 +402,32 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public boolean isEventDrivenSupported() {
- readLock.lock();
- try {
- return this.eventDrivenSupported;
- } finally {
- readLock.unlock();
- }
+ return this.eventDrivenSupported;
}
/**
* Updates the Scheduling Strategy used for this Processor
*
- * @param schedulingStrategy strategy
+ * @param schedulingStrategy
+ * strategy
*
- * @throws IllegalArgumentException if the SchedulingStrategy is not not applicable for this Processor
+ * @throws IllegalArgumentException
+ * if the SchedulingStrategy is not not applicable for this
+ * Processor
*/
@Override
public void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy) {
- writeLock.lock();
- try {
- if (schedulingStrategy == SchedulingStrategy.EVENT_DRIVEN && !eventDrivenSupported) {
- // not valid. Just ignore it. We don't throw an Exception because if a developer changes a Processor so that
- // it no longer supports EventDriven mode, we don't want the app to fail to startup if it was already in Event-Driven
- // Mode. Instead, we will simply leave it in Timer-Driven mode
- return;
- }
-
- this.schedulingStrategy = schedulingStrategy;
- setIsolated(schedulingStrategy == SchedulingStrategy.PRIMARY_NODE_ONLY);
- } finally {
- writeLock.unlock();
+ if (schedulingStrategy == SchedulingStrategy.EVENT_DRIVEN && !eventDrivenSupported) {
+ // not valid. Just ignore it. We don't throw an Exception because if
+ // a developer changes a Processor so that
+ // it no longer supports EventDriven mode, we don't want the app to
+ // fail to startup if it was already in Event-Driven
+ // Mode. Instead, we will simply leave it in Timer-Driven mode
+ return;
}
+
+ this.schedulingStrategy = schedulingStrategy;
+ setIsolated(schedulingStrategy == SchedulingStrategy.PRIMARY_NODE_ONLY);
}
/**
@@ -426,12 +435,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
*/
@Override
public SchedulingStrategy getSchedulingStrategy() {
- readLock.lock();
- try {
- return this.schedulingStrategy;
- } finally {
- readLock.unlock();
- }
+ return this.schedulingStrategy;
}
@Override
@@ -441,63 +445,51 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void setScheduldingPeriod(final String schedulingPeriod) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
+ }
- switch (schedulingStrategy) {
- case CRON_DRIVEN: {
- try {
- new CronExpression(schedulingPeriod);
- } catch (final Exception e) {
- throw new IllegalArgumentException("Scheduling Period is not a valid cron expression: " + schedulingPeriod);
- }
- }
- break;
- case PRIMARY_NODE_ONLY:
- case TIMER_DRIVEN: {
- final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod), TimeUnit.NANOSECONDS);
- if (schedulingNanos < 0) {
- throw new IllegalArgumentException("Scheduling Period must be positive");
- }
- this.schedulingNanos.set(Math.max(MINIMUM_SCHEDULING_NANOS, schedulingNanos));
- }
- break;
- case EVENT_DRIVEN:
- default:
- return;
+ switch (schedulingStrategy) {
+ case CRON_DRIVEN: {
+ try {
+ new CronExpression(schedulingPeriod);
+ } catch (final Exception e) {
+ throw new IllegalArgumentException(
+ "Scheduling Period is not a valid cron expression: " + schedulingPeriod);
}
-
- this.schedulingPeriod.set(schedulingPeriod);
- } finally {
- writeLock.unlock();
}
+ break;
+ case PRIMARY_NODE_ONLY:
+ case TIMER_DRIVEN: {
+ final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod),
+ TimeUnit.NANOSECONDS);
+ if (schedulingNanos < 0) {
+ throw new IllegalArgumentException("Scheduling Period must be positive");
+ }
+ this.schedulingNanos.set(Math.max(MINIMUM_SCHEDULING_NANOS, schedulingNanos));
+ }
+ break;
+ case EVENT_DRIVEN:
+ default:
+ return;
+ }
+
+ this.schedulingPeriod.set(schedulingPeriod);
}
@Override
public long getRunDuration(final TimeUnit timeUnit) {
- readLock.lock();
- try {
- return timeUnit.convert(this.runNanos, TimeUnit.NANOSECONDS);
- } finally {
- readLock.unlock();
- }
+ return timeUnit.convert(this.runNanos, TimeUnit.NANOSECONDS);
}
@Override
public void setRunDuration(final long duration, final TimeUnit timeUnit) {
- writeLock.lock();
- try {
- if (duration < 0) {
- throw new IllegalArgumentException("Run Duration must be non-negative value; cannot set to " + timeUnit.toSeconds(duration) + " seconds");
- }
-
- this.runNanos = timeUnit.toNanos(duration);
- } finally {
- writeLock.unlock();
+ if (duration < 0) {
+ throw new IllegalArgumentException("Run Duration must be non-negative value; cannot set to "
+ + timeUnit.toSeconds(duration) + " seconds");
}
+
+ this.runNanos = timeUnit.toNanos(duration);
}
@Override
@@ -512,32 +504,32 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void setYieldPeriod(final String yieldPeriod) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- final long yieldMillis = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.MILLISECONDS);
- if (yieldMillis < 0) {
- throw new IllegalArgumentException("Yield duration must be positive");
- }
- this.yieldPeriod.set(yieldPeriod);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ final long yieldMillis = FormatUtils.getTimeDuration(requireNonNull(yieldPeriod), TimeUnit.MILLISECONDS);
+ if (yieldMillis < 0) {
+ throw new IllegalArgumentException("Yield duration must be positive");
+ }
+ this.yieldPeriod.set(yieldPeriod);
}
/**
- * Causes the processor not to be scheduled for some period of time. This duration can be obtained and set via the {@link #getYieldPeriod(TimeUnit)} and {@link #setYieldPeriod(long, TimeUnit)}
- * methods.
+ * Causes the processor not to be scheduled for some period of time. This
+ * duration can be obtained and set via the
+ * {@link #getYieldPeriod(TimeUnit)} and
+ * {@link #setYieldPeriod(long, TimeUnit)} methods.
*/
@Override
public void yield() {
final long yieldMillis = getYieldPeriod(TimeUnit.MILLISECONDS);
yield(yieldMillis, TimeUnit.MILLISECONDS);
- final String yieldDuration = (yieldMillis > 1000) ? (yieldMillis / 1000) + " seconds" : yieldMillis + " milliseconds";
- LoggerFactory.getLogger(processor.getClass()).debug("{} has chosen to yield its resources; will not be scheduled to run again for {}", processor, yieldDuration);
+ final String yieldDuration = (yieldMillis > 1000) ? (yieldMillis / 1000) + " seconds"
+ : yieldMillis + " milliseconds";
+ LoggerFactory.getLogger(processor.getClass()).debug(
+ "{} has chosen to yield its resources; will not be scheduled to run again for {}", processor,
+ yieldDuration);
}
@Override
@@ -549,7 +541,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return the number of milliseconds since Epoch at which time this processor is to once again be scheduled.
+ * @return the number of milliseconds since Epoch at which time this
+ * processor is to once again be scheduled.
*/
@Override
public long getYieldExpiration() {
@@ -568,42 +561,36 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void setPenalizationPeriod(final String penalizationPeriod) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- final long penalizationMillis = FormatUtils.getTimeDuration(requireNonNull(penalizationPeriod), TimeUnit.MILLISECONDS);
- if (penalizationMillis < 0) {
- throw new IllegalArgumentException("Penalization duration must be positive");
- }
- this.penalizationPeriod.set(penalizationPeriod);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
+ }
+ final long penalizationMillis = FormatUtils.getTimeDuration(requireNonNull(penalizationPeriod),
+ TimeUnit.MILLISECONDS);
+ if (penalizationMillis < 0) {
+ throw new IllegalArgumentException("Penalization duration must be positive");
}
+ this.penalizationPeriod.set(penalizationPeriod);
}
/**
- * Determines the number of concurrent tasks that may be running for this processor.
+ * Determines the number of concurrent tasks that may be running for this
+ * processor.
*
- * @param taskCount a number of concurrent tasks this processor may have running
- * @throws IllegalArgumentException if the given value is less than 1
+ * @param taskCount
+ * a number of concurrent tasks this processor may have running
+ * @throws IllegalArgumentException
+ * if the given value is less than 1
*/
@Override
public void setMaxConcurrentTasks(final int taskCount) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
- }
- if (taskCount < 1 && getSchedulingStrategy() != SchedulingStrategy.EVENT_DRIVEN) {
- throw new IllegalArgumentException();
- }
- if (!triggeredSerially) {
- concurrentTaskCount.set(taskCount);
- }
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
+ }
+ if (taskCount < 1 && getSchedulingStrategy() != SchedulingStrategy.EVENT_DRIVEN) {
+ throw new IllegalArgumentException();
+ }
+ if (!triggeredSerially) {
+ concurrentTaskCount.set(taskCount);
}
}
@@ -613,7 +600,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return the number of tasks that may execute concurrently for this processor
+ * @return the number of tasks that may execute concurrently for this
+ * processor
*/
@Override
public int getMaxConcurrentTasks() {
@@ -633,13 +621,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public Set<Connection> getConnections() {
final Set<Connection> allConnections = new HashSet<>();
- readLock.lock();
- try {
- for (final Set<Connection> connectionSet : connections.values()) {
- allConnections.addAll(connectionSet);
- }
- } finally {
- readLock.unlock();
+ for (final Set<Connection> connectionSet : connections.values()) {
+ allConnections.addAll(connectionSet);
}
return allConnections;
@@ -652,14 +635,9 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public Set<Connection> getConnections(final Relationship relationship) {
- final Set<Connection> applicableConnections;
- readLock.lock();
- try {
- applicableConnections = connections.get(relationship);
- } finally {
- readLock.unlock();
- }
- return (applicableConnections == null) ? Collections.<Connection> emptySet() : Collections.unmodifiableSet(applicableConnections);
+ Set<Connection> applicableConnections = connections.get(relationship);
+ return (applicableConnections == null) ? Collections.<Connection> emptySet()
+ : Collections.unmodifiableSet(applicableConnections);
}
@Override
@@ -667,52 +645,52 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
Objects.requireNonNull(connection, "connection cannot be null");
if (!connection.getSource().equals(this) && !connection.getDestination().equals(this)) {
- throw new IllegalStateException("Cannot a connection to a ProcessorNode for which the ProcessorNode is neither the Source nor the Destination");
+ throw new IllegalStateException(
+ "Cannot a connection to a ProcessorNode for which the ProcessorNode is neither the Source nor the Destination");
}
- writeLock.lock();
- try {
- List<Connection> updatedIncoming = null;
- if (connection.getDestination().equals(this)) {
- // don't add the connection twice. This may occur if we have a self-loop because we will be told
- // to add the connection once because we are the source and again because we are the destination.
- final List<Connection> incomingConnections = incomingConnectionsRef.get();
- updatedIncoming = new ArrayList<>(incomingConnections);
- if (!updatedIncoming.contains(connection)) {
- updatedIncoming.add(connection);
- }
+ List<Connection> updatedIncoming = null;
+ if (connection.getDestination().equals(this)) {
+ // don't add the connection twice. This may occur if we have a
+ // self-loop because we will be told
+ // to add the connection once because we are the source and again
+ // because we are the destination.
+ final List<Connection> incomingConnections = incomingConnectionsRef.get();
+ updatedIncoming = new ArrayList<>(incomingConnections);
+ if (!updatedIncoming.contains(connection)) {
+ updatedIncoming.add(connection);
}
+ }
- if (connection.getSource().equals(this)) {
- // don't add the connection twice. This may occur if we have a self-loop because we will be told
- // to add the connection once because we are the source and again because we are the destination.
- if (!destinations.containsKey(connection)) {
- for (final Relationship relationship : connection.getRelationships()) {
- final Relationship rel = getRelationship(relationship.getName());
- Set<Connection> set = connections.get(rel);
- if (set == null) {
- set = new HashSet<>();
- connections.put(rel, set);
- }
+ if (connection.getSource().equals(this)) {
+ // don't add the connection twice. This may occur if we have a
+ // self-loop because we will be told
+ // to add the connection once because we are the source and again
+ // because we are the destination.
+ if (!destinations.containsKey(connection)) {
+ for (final Relationship relationship : connection.getRelationships()) {
+ final Relationship rel = getRelationship(relationship.getName());
+ Set<Connection> set = connections.get(rel);
+ if (set == null) {
+ set = new HashSet<>();
+ connections.put(rel, set);
+ }
- set.add(connection);
+ set.add(connection);
- destinations.put(connection, connection.getDestination());
- }
+ destinations.put(connection, connection.getDestination());
+ }
- final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
- if (autoTerminated != null) {
- autoTerminated.removeAll(connection.getRelationships());
- this.undefinedRelationshipsToTerminate.set(autoTerminated);
- }
+ final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
+ if (autoTerminated != null) {
+ autoTerminated.removeAll(connection.getRelationships());
+ this.undefinedRelationshipsToTerminate.set(autoTerminated);
}
}
+ }
- if (updatedIncoming != null) {
- incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
- }
- } finally {
- writeLock.unlock();
+ if (updatedIncoming != null) {
+ incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
}
}
@@ -724,74 +702,68 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void updateConnection(final Connection connection) throws IllegalStateException {
if (requireNonNull(connection).getSource().equals(this)) {
- writeLock.lock();
- try {
- //
- // update any relationships
- //
- // first check if any relations were removed.
- final List<Relationship> existingRelationships = new ArrayList<>();
- for (final Map.Entry<Relationship, Set<Connection>> entry : connections.entrySet()) {
- if (entry.getValue().contains(connection)) {
- existingRelationships.add(entry.getKey());
- }
+ // update any relationships
+ //
+ // first check if any relations were removed.
+ final List<Relationship> existingRelationships = new ArrayList<>();
+ for (final Map.Entry<Relationship, Set<Connection>> entry : connections.entrySet()) {
+ if (entry.getValue().contains(connection)) {
+ existingRelationships.add(entry.getKey());
}
+ }
- for (final Relationship rel : connection.getRelationships()) {
- if (!existingRelationships.contains(rel)) {
- // relationship was removed. Check if this is legal.
- final Set<Connection> connectionsForRelationship = getConnections(rel);
- if (connectionsForRelationship != null && connectionsForRelationship.size() == 1 && this.isRunning() && !isAutoTerminated(rel) && getRelationships().contains(rel)) {
- // if we are running and we do not terminate undefined relationships and this is the only
- // connection that defines the given relationship, and that relationship is required,
- // then it is not legal to remove this relationship from this connection.
- throw new IllegalStateException("Cannot remove relationship " + rel.getName() + " from Connection because doing so would invalidate Processor "
- + this + ", which is currently running");
- }
+ for (final Relationship rel : connection.getRelationships()) {
+ if (!existingRelationships.contains(rel)) {
+ // relationship was removed. Check if this is legal.
+ final Set<Connection> connectionsForRelationship = getConnections(rel);
+ if (connectionsForRelationship != null && connectionsForRelationship.size() == 1 && this.isRunning()
+ && !isAutoTerminated(rel) && getRelationships().contains(rel)) {
+ // if we are running and we do not terminate undefined
+ // relationships and this is the only
+ // connection that defines the given relationship, and
+ // that relationship is required,
+ // then it is not legal to remove this relationship from
+ // this connection.
+ throw new IllegalStateException("Cannot remove relationship " + rel.getName()
+ + " from Connection because doing so would invalidate Processor " + this
+ + ", which is currently running");
}
}
+ }
- // remove the connection from any list that currently contains
- for (final Set<Connection> list : connections.values()) {
- list.remove(connection);
- }
+ // remove the connection from any list that currently contains
+ for (final Set<Connection> list : connections.values()) {
+ list.remove(connection);
+ }
- // add the connection in for all relationships listed.
- for (final Relationship rel : connection.getRelationships()) {
- Set<Connection> set = connections.get(rel);
- if (set == null) {
- set = new HashSet<>();
- connections.put(rel, set);
- }
- set.add(connection);
+ // add the connection in for all relationships listed.
+ for (final Relationship rel : connection.getRelationships()) {
+ Set<Connection> set = connections.get(rel);
+ if (set == null) {
+ set = new HashSet<>();
+ connections.put(rel, set);
}
+ set.add(connection);
+ }
- // update to the new destination
- destinations.put(connection, connection.getDestination());
+ // update to the new destination
+ destinations.put(connection, connection.getDestination());
- final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
- if (autoTerminated != null) {
- autoTerminated.removeAll(connection.getRelationships());
- this.undefinedRelationshipsToTerminate.set(autoTerminated);
- }
- } finally {
- writeLock.unlock();
+ final Set<Relationship> autoTerminated = this.undefinedRelationshipsToTerminate.get();
+ if (autoTerminated != null) {
+ autoTerminated.removeAll(connection.getRelationships());
+ this.undefinedRelationshipsToTerminate.set(autoTerminated);
}
}
if (connection.getDestination().equals(this)) {
- writeLock.lock();
- try {
- // update our incoming connections -- we can just remove & re-add the connection to
- // update the list.
- final List<Connection> incomingConnections = incomingConnectionsRef.get();
- final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
- updatedIncoming.remove(connection);
- updatedIncoming.add(connection);
- incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
- } finally {
- writeLock.unlock();
- }
+ // update our incoming connections -- we can just remove & re-add
+ // the connection to update the list.
+ final List<Connection> incomingConnections = incomingConnectionsRef.get();
+ final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
+ updatedIncoming.remove(connection);
+ updatedIncoming.add(connection);
+ incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
}
}
@@ -803,45 +775,39 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
for (final Relationship relationship : connection.getRelationships()) {
final Set<Connection> connectionsForRelationship = getConnections(relationship);
if ((connectionsForRelationship == null || connectionsForRelationship.size() <= 1) && isRunning()) {
- throw new IllegalStateException("This connection cannot be removed because its source is running and removing it will invalidate this processor");
+ throw new IllegalStateException(
+ "This connection cannot be removed because its source is running and removing it will invalidate this processor");
}
}
- writeLock.lock();
- try {
- for (final Set<Connection> connectionList : this.connections.values()) {
- connectionList.remove(connection);
- }
-
- connectionRemoved = (destinations.remove(connection) != null);
- } finally {
- writeLock.unlock();
+ for (final Set<Connection> connectionList : this.connections.values()) {
+ connectionList.remove(connection);
}
+
+ connectionRemoved = (destinations.remove(connection) != null);
}
if (connection.getDestination().equals(this)) {
- writeLock.lock();
- try {
- final List<Connection> incomingConnections = incomingConnectionsRef.get();
- if (incomingConnections.contains(connection)) {
- final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
- updatedIncoming.remove(connection);
- incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
- return;
- }
- } finally {
- writeLock.unlock();
+ final List<Connection> incomingConnections = incomingConnectionsRef.get();
+ if (incomingConnections.contains(connection)) {
+ final List<Connection> updatedIncoming = new ArrayList<>(incomingConnections);
+ updatedIncoming.remove(connection);
+ incomingConnectionsRef.set(Collections.unmodifiableList(updatedIncoming));
+ return;
}
}
if (!connectionRemoved) {
- throw new IllegalArgumentException("Cannot remove a connection from a ProcessorNode for which the ProcessorNode is not the Source");
+ throw new IllegalArgumentException(
+ "Cannot remove a connection from a ProcessorNode for which the ProcessorNode is not the Source");
}
}
/**
- * @param relationshipName name
- * @return the relationship for this nodes processor for the given name or creates a new relationship for the given name
+ * @param relationshipName
+ * name
+ * @return the relationship for this nodes processor for the given name or
+ * creates a new relationship for the given name
*/
@Override
public Relationship getRelationship(final String relationshipName) {
@@ -868,59 +834,45 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
/**
- * @return the Set of destination processors for all relationships excluding any destinations that are this processor itself (self-loops)
+ * @return the Set of destination processors for all relationships excluding
+ * any destinations that are this processor itself (self-loops)
*/
public Set<Connectable> getDestinations() {
final Set<Connectable> nonSelfDestinations = new HashSet<>();
- readLock.lock();
- try {
- for (final Connectable connectable : destinations.values()) {
- if (connectable != this) {
- nonSelfDestinations.add(connectable);
- }
+ for (final Connectable connectable : destinations.values()) {
+ if (connectable != this) {
+ nonSelfDestinations.add(connectable);
}
- } finally {
- readLock.unlock();
}
return nonSelfDestinations;
}
public Set<Connectable> getDestinations(final Relationship relationship) {
- readLock.lock();
- try {
- final Set<Connectable> destinationSet = new HashSet<>();
- final Set<Connection> relationshipConnections = connections.get(relationship);
- if (relationshipConnections != null) {
- for (final Connection connection : relationshipConnections) {
- destinationSet.add(destinations.get(connection));
- }
+ final Set<Connectable> destinationSet = new HashSet<>();
+ final Set<Connection> relationshipConnections = connections.get(relationship);
+ if (relationshipConnections != null) {
+ for (final Connection connection : relationshipConnections) {
+ destinationSet.add(destinations.get(connection));
}
- return destinationSet;
- } finally {
- readLock.unlock();
}
+ return destinationSet;
}
public Set<Relationship> getUndefinedRelationships() {
final Set<Relationship> undefined = new HashSet<>();
- readLock.lock();
- try {
- final Set<Relationship> relationships;
- try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
- relationships = processor.getRelationships();
- }
+ final Set<Relationship> relationships;
+ try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
+ relationships = processor.getRelationships();
+ }
- if (relationships == null) {
- return undefined;
- }
- for (final Relationship relation : relationships) {
- final Set<Connection> connectionSet = this.connections.get(relation);
- if (connectionSet == null || connectionSet.isEmpty()) {
- undefined.add(relation);
- }
+ if (relationships == null) {
+ return undefined;
+ }
+ for (final Relationship relation : relationships) {
+ final Set<Connection> connectionSet = this.connections.get(relation);
+ if (connectionSet == null || connectionSet.isEmpty()) {
+ undefined.add(relation);
}
- } finally {
- readLock.unlock();
}
return undefined;
}
@@ -928,36 +880,22 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
/**
* Determines if the given node is a destination for this node
*
- * @param node node
+ * @param node
+ * node
* @return true if is a direct destination node; false otherwise
*/
boolean isRelated(final ProcessorNode node) {
- readLock.lock();
- try {
- return this.destinations.containsValue(node);
- } finally {
- readLock.unlock();
- }
+ return this.destinations.containsValue(node);
}
@Override
public boolean isRunning() {
- readLock.lock();
- try {
- return getScheduledState().equals(ScheduledState.RUNNING) || processScheduler.getActiveThreadCount(this) > 0;
- } finally {
- readLock.unlock();
- }
+ return getScheduledState().equals(ScheduledState.RUNNING) || processScheduler.getActiveThreadCount(this) > 0;
}
@Override
public int getActiveThreadCount() {
- readLock.lock();
- try {
- return processScheduler.getActiveThreadCount(this);
- } finally {
- readLock.unlock();
- }
+ return processScheduler.getActiveThreadCount(this);
}
List<Connection> getIncomingNonLoopConnections() {
@@ -974,14 +912,12 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public boolean isValid() {
- readLock.lock();
try {
- final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData());
+ final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(),
+ getAnnotationData());
final Collection<ValidationResult> validationResults;
- try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
- validationResults = getProcessor().validate(validationContext);
- }
+ validationResults = getProcessor().validate(validationContext);
for (final ValidationResult result : validationResults) {
if (!result.isValid()) {
@@ -996,36 +932,34 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
switch (getInputRequirement()) {
- case INPUT_ALLOWED:
- break;
- case INPUT_FORBIDDEN: {
- if (!getIncomingNonLoopConnections().isEmpty()) {
- return false;
- }
- break;
+ case INPUT_ALLOWED:
+ break;
+ case INPUT_FORBIDDEN: {
+ if (!getIncomingNonLoopConnections().isEmpty()) {
+ return false;
}
- case INPUT_REQUIRED: {
- if (getIncomingNonLoopConnections().isEmpty()) {
- return false;
- }
- break;
+ break;
+ }
+ case INPUT_REQUIRED: {
+ if (getIncomingNonLoopConnections().isEmpty()) {
+ return false;
}
+ break;
+ }
}
} catch (final Throwable t) {
+ LOG.warn("Failed during validation", t);
return false;
- } finally {
- readLock.unlock();
}
-
return true;
}
@Override
public Collection<ValidationResult> getValidationErrors() {
final List<ValidationResult> results = new ArrayList<>();
- readLock.lock();
try {
- final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(), getAnnotationData());
+ final ValidationContext validationContext = validationContextFactory.newValidationContext(getProperties(),
+ getAnnotationData());
final Collection<ValidationResult> validationResults;
try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
@@ -1041,43 +975,37 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
for (final Relationship relationship : getUndefinedRelationships()) {
if (!isAutoTerminated(relationship)) {
final ValidationResult error = new ValidationResult.Builder()
- .explanation("Relationship '" + relationship.getName() + "' is not connected to any component and is not auto-terminated")
- .subject("Relationship " + relationship.getName())
- .valid(false)
- .build();
+ .explanation("Relationship '" + relationship.getName()
+ + "' is not connected to any component and is not auto-terminated")
+ .subject("Relationship " + relationship.getName()).valid(false).build();
results.add(error);
}
}
switch (getInputRequirement()) {
- case INPUT_ALLOWED:
- break;
- case INPUT_FORBIDDEN: {
- final int incomingConnCount = getIncomingNonLoopConnections().size();
- if (incomingConnCount != 0) {
- results.add(new ValidationResult.Builder()
- .explanation("Processor does not allow upstream connections but currently has " + incomingConnCount)
- .subject("Upstream Connections")
- .valid(false)
- .build());
- }
- break;
+ case INPUT_ALLOWED:
+ break;
+ case INPUT_FORBIDDEN: {
+ final int incomingConnCount = getIncomingNonLoopConnections().size();
+ if (incomingConnCount != 0) {
+ results.add(new ValidationResult.Builder().explanation(
+ "Processor does not allow upstream connections but currently has " + incomingConnCount)
+ .subject("Upstream Connections").valid(false).build());
}
- case INPUT_REQUIRED: {
- if (getIncomingNonLoopConnections().isEmpty()) {
- results.add(new ValidationResult.Builder()
+ break;
+ }
+ case INPUT_REQUIRED: {
+ if (getIncomingNonLoopConnections().isEmpty()) {
+ results.add(new ValidationResult.Builder()
.explanation("Processor requires an upstream connection but currently has none")
- .subject("Upstream Connections")
- .valid(false)
- .build());
- }
- break;
+ .subject("Upstream Connections").valid(false).build());
}
+ break;
+ }
}
} catch (final Throwable t) {
- results.add(new ValidationResult.Builder().explanation("Failed to run validation due to " + t.toString()).valid(false).build());
- } finally {
- readLock.unlock();
+ results.add(new ValidationResult.Builder().explanation("Failed to run validation due to " + t.toString())
+ .valid(false).build());
}
return results;
}
@@ -1090,7 +1018,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
/**
* Establishes node equality (based on the processor's identifier)
*
- * @param other node
+ * @param other
+ * node
* @return true if equal
*/
@Override
@@ -1128,18 +1057,15 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void setProcessGroup(final ProcessGroup group) {
- writeLock.lock();
- try {
- this.processGroup.set(group);
- } finally {
- writeLock.unlock();
- }
+ this.processGroup.set(group);
}
@Override
public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) {
try (final NarCloseable narCloseable = NarCloseable.withNarLoader()) {
processor.onTrigger(context, sessionFactory);
+ } catch (Exception ex) {
+ ex.printStackTrace();
}
}
@@ -1149,25 +1075,12 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
}
@Override
- public void setScheduledState(final ScheduledState scheduledState) {
- this.scheduledState.set(scheduledState);
- if (!scheduledState.equals(ScheduledState.RUNNING)) { // if user stops processor, clear yield expiration
- yieldExpiration.set(0L);
- }
- }
-
- @Override
public void setAnnotationData(final String data) {
- writeLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException("Cannot set AnnotationData while processor is running");
- }
-
- this.annotationData.set(data);
- } finally {
- writeLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException("Cannot set AnnotationData while processor is running");
}
+
+ this.annotationData.set(data);
}
@Override
@@ -1187,75 +1100,55 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void verifyCanDelete(final boolean ignoreConnections) {
- readLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException(this + " is running");
- }
+ if (isRunning()) {
+ throw new IllegalStateException(this + " is running");
+ }
- if (!ignoreConnections) {
- for (final Set<Connection> connectionSet : connections.values()) {
- for (final Connection connection : connectionSet) {
- connection.verifyCanDelete();
- }
+ if (!ignoreConnections) {
+ for (final Set<Connection> connectionSet : connections.values()) {
+ for (final Connection connection : connectionSet) {
+ connection.verifyCanDelete();
}
+ }
- for (final Connection connection : incomingConnectionsRef.get()) {
- if (connection.getSource().equals(this)) {
- connection.verifyCanDelete();
- } else {
- throw new IllegalStateException(this + " is the destination of another component");
- }
+ for (final Connection connection : incomingConnectionsRef.get()) {
+ if (connection.getSource().equals(this)) {
+ connection.verifyCanDelete();
+ } else {
+ throw new IllegalStateException(this + " is the destination of another component");
}
}
- } finally {
- readLock.unlock();
}
}
@Override
public void verifyCanStart() {
- readLock.lock();
- try {
- switch (getScheduledState()) {
- case DISABLED:
- throw new IllegalStateException(this + " cannot be started because it is disabled");
- case RUNNING:
- throw new IllegalStateException(this + " cannot be started because it is already running");
- case STOPPED:
- break;
- }
- verifyNoActiveThreads();
-
- if (!isValid()) {
- throw new IllegalStateException(this + " is not in a valid state");
- }
- } finally {
- readLock.unlock();
- }
+ this.verifyCanStart(null);
}
@Override
public void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences) {
- switch (getScheduledState()) {
- case DISABLED:
- throw new IllegalStateException(this + " cannot be started because it is disabled");
- case RUNNING:
- throw new IllegalStateException(this + " cannot be started because it is already running");
- case STOPPED:
- break;
+ if (this.getScheduledState() == ScheduledState.RUNNING) {
+ throw new IllegalStateException(this + " cannot be started because it is already running");
}
+
verifyNoActiveThreads();
- final Set<String> ids = new HashSet<>();
- for (final ControllerServiceNode node : ignoredReferences) {
- ids.add(node.getIdentifier());
- }
+ if (ignoredReferences != null) {
+ final Set<String> ids = new HashSet<>();
+ for (final ControllerServiceNode node : ignoredReferences) {
+ ids.add(node.getIdentifier());
+ }
- final Collection<ValidationResult> validationResults = getValidationErrors(ids);
- for (final ValidationResult result : validationResults) {
- if (!result.isValid()) {
- throw new IllegalStateException(this + " cannot be started because it is not valid: " + result);
+ final Collection<ValidationResult> validationResults = getValidationErrors(ids);
+ for (final ValidationResult result : validationResults) {
+ if (!result.isValid()) {
+ throw new IllegalStateException(this + " cannot be started because it is not valid: " + result);
+ }
+ }
+ } else {
+ if (!isValid()) {
+ throw new IllegalStateException(this + " is not in a valid state");
}
}
}
@@ -1269,41 +1162,26 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
@Override
public void verifyCanUpdate() {
- readLock.lock();
- try {
- if (isRunning()) {
- throw new IllegalStateException(this + " is not stopped");
- }
- } finally {
- readLock.unlock();
+ if (isRunning()) {
+ throw new IllegalStateException(this + " is not stopped");
}
}
@Override
public void verifyCanEnable() {
- readLock.lock();
- try {
- if (getScheduledState() != ScheduledState.DISABLED) {
- throw new IllegalStateException(this + " is not disabled");
- }
-
- verifyNoActiveThreads();
- } finally {
- readLock.unlock();
+ if (getScheduledState() != ScheduledState.DISABLED) {
+ throw new IllegalStateException(this + " is not disabled");
}
+
+ verifyNoActiveThreads();
}
@Override
public void verifyCanDisable() {
- readLock.lock();
- try {
- if (getScheduledState() != ScheduledState.STOPPED) {
- throw new IllegalStateException(this + " is not stopped");
- }
- verifyNoActiveThreads();
- } finally {
- readLock.unlock();
+ if (getScheduledState() != ScheduledState.STOPPED) {
+ throw new IllegalStateException(this + " is not stopped");
}
+ verifyNoActiveThreads();
}
@Override
@@ -1324,4 +1202,195 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
}
+
+ /**
+ * Will idempotently start the processor using the following sequence: <i>
+ * <ul>
+ * <li>Validate Processor's state (e.g., PropertyDescriptors,
+ * ControllerServices etc.)</li>
+ * <li>Transition (atomically) Processor's scheduled state form STOPPED to
+ * STARTING. If the above state transition succeeds, then execute the start
+ * task (asynchronously) which will be re-tried until @OnScheduled is
+ * executed successfully and "schedulingAgentCallback' is invoked, or until
+ * STOP operation is initiated on this processor. If state transition fails
+ * it means processor is already being started and WARN message will be
+ * logged explaining it.</li>
+ * </ul>
+ * </i>
+ * <p>
+ * Any exception thrown while invoking operations annotated with @OnSchedule
+ * will be caught and logged after which @OnUnscheduled operation will be
+ * invoked (quietly) and the start sequence will be repeated (re-try) after
+ * delay provided by 'administrativeYieldMillis'.
+ * </p>
+ * <p>
+ * Upon successful completion of start sequence (@OnScheduled ->
+ * 'schedulingAgentCallback') the attempt will be made to transition
+ * processor's scheduling state to RUNNING at which point processor is
+ * considered to be fully started and functioning. If upon successful
+ * invocation of @OnScheduled operation the processor can not be
+ * transitioned to RUNNING state (e.g., STOP operation was invoked on the
+ * processor while it's @OnScheduled operation was executing), the
+ * processor's @OnUnscheduled operation will be invoked and its scheduling
+ * state will be set to STOPPED at which point the processor is considered
+ * to be fully stopped.
+ * </p>
+ */
+ @Override
+ public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler,
+ final long administrativeYieldMillis, final T processContext, final Runnable schedulingAgentCallback) {
+ if (!this.isValid()) {
+ throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors());
+ }
+
+ if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { // will ensure that the Processor represented by this node can only be started once
+ final Runnable startProcRunnable = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ SchedulingContext schedulingContext = new StandardSchedulingContext(processContext, getControllerServiceProvider(),
+ StandardProcessorNode.this, processContext.getStateManager());
+ invokeOnScheduleAsync(taskScheduler, schedulingContext);
+ if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) {
+ schedulingAgentCallback.run(); // callback provided by StandardProcessScheduler to essentially initiate component's onTrigger() cycle
+ } else { // can only happen if stopProcessor was called before service was transitioned to RUNNING state
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
+ scheduledState.set(ScheduledState.STOPPED);
+ }
+ } catch (final Exception e) {
+ final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e;
+ final ProcessorLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor);
+
+ 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);
+ LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause);
+
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
+ if (scheduledState.get() != ScheduledState.STOPPING) { // make sure we only continue retry loop if STOP action wasn't initiated
+ taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS);
+ } else {
+ scheduledState.set(ScheduledState.STOPPED);
+ }
+ }
+ }
+ };
+ taskScheduler.execute(startProcRunnable);
+ } else {
+ LOG.warn("Can not start Processor since it's already in the process of being started");
+ }
+ }
+
+ /**
+ * 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
+ * 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
+ * stop was called), the attempt will be made to transition processor's
+ * scheduled state from STARTING to STOPPING which will allow
+ * {@link #start(ScheduledExecutorService, long, ProcessContext, Runnable)}
+ * method to initiate processor's shutdown upon exiting @OnScheduled
+ * operation, otherwise the processor's scheduled state will remain
+ * unchanged ensuring that multiple calls to this method are idempotent.
+ * </p>
+ */
+ @Override
+ public <T extends ProcessContext & ControllerServiceLookup> void stop(final ScheduledExecutorService scheduler,
+ final T processContext, final Callable<Boolean> activeThreadMonitorCallback) {
+ if (this.scheduledState.compareAndSet(ScheduledState.RUNNING, ScheduledState.STOPPING)) { // will ensure that the Processor represented by this node can only be stopped once
+ final Runnable stopProcRunnable = new Runnable() {
+ @Override
+ public void run() {
+ try {
+ if (activeThreadMonitorCallback.call()) {
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext);
+ ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext);
+ scheduledState.set(ScheduledState.STOPPED);
+ } else {
+ scheduler.schedule(this, 100, TimeUnit.MILLISECONDS);
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed while shutting down processor " + processor, e);
+ }
+ }
+ };
+ scheduler.execute(stopProcRunnable);
+ } else {
+ /*
+ * We do compareAndSet() instead of set() to ensure that Processor
+ * stoppage is handled consistently including a condition where
+ * Processor never got a chance to transition to RUNNING state
+ * before stop() was called. If that happens the stop processor
+ * routine will be initiated in start() method, otherwise the IF
+ * part will handle the stop processor routine.
+ */
+ this.scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.STOPPING);
+ }
+ }
+
+ /**
+ * Will invoke processor's methods annotated with @OnSchedule asynchronously
+ * to ensure that it could be interrupted if stop action was initiated on
+ * the processor that may be sitting in the infinitely blocking @OnSchedule
+ * operation. While this approach paves the way for further enhancements
+ * related to managing processor'slife-cycle operation at the moment the
+ * interrupt will not happen automatically. This is primarily to preserve
+ * the existing behavior or the NiFi where stop operation can only be
+ * invoked once the processor is started. Unfortunately that could mean that
+ * the processor may be blocking indefinitely in the @Oncheduled call. To
+ * deal with that a new NiFi property has been introduced
+ * <i>nifi.processor.start.timeout</i> which allows one to set the time (in
+ * milliseconds) of how long to wait before canceling the @OnScheduled task
+ * allowing processor's stop sequence to proceed. The default value for this
+ * property is {@link Long#MAX_VALUE}.
+ * <p>
+ * NOTE: Canceling the task does not guarantee that the task will actually
+ * completes (successfully or otherwise), since cancellation of the task
+ * will issue a simple Thread.interrupt(). However code inside
+ * of @OnScheduled operation is written purely and will ignore thread
+ * interrupts you may end up with runaway thread which may eventually
+ * require NiFi reboot. In any event, the above explanation will be logged
+ * (WARN) informing a user so further actions could be taken.
+ * </p>
+ */
+ private void invokeOnScheduleAsync(ScheduledExecutorService taskScheduler, final SchedulingContext schedulingContext) throws ExecutionException {
+ Future<Void> executionResult = taskScheduler.submit(new Callable<Void>() {
+ @SuppressWarnings("deprecation")
+ @Override
+ public Void call() throws Exception {
+ ReflectionUtils.invokeMethodsWithAnnotations(OnScheduled.class, org.apache.nifi.processor.annotation.OnScheduled.class, processor, schedulingContext);
+ return null;
+ }
+ });
+
+ long onScheduleTimeout = Long.parseLong(NiFiProperties.getInstance()
+ .getProperty(NiFiProperties.PROCESSOR_START_TIMEOUT, String.valueOf(Long.MAX_VALUE)));
+ try {
+ executionResult.get(onScheduleTimeout, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ LOG.warn("Thread was interrupted while waiting for processor '" + this.processor.getClass().getSimpleName()
+ + "' @OnSchedule operation to finish.");
+ Thread.currentThread().interrupt();
+ } catch (TimeoutException e) {
+ executionResult.cancel(true);
+ LOG.warn("Timed out while waiting for the task executing @OnSchedule operation for '"
+ + 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 @OnSchedule method may "
+ + "have been written to ignore interrupts which may result in runaway thread which 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.");
+ }
+ }
}
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/AbstractSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
new file mode 100644
index 0000000..b931c64
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/AbstractSchedulingAgent.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.controller.scheduling;
+
+import org.apache.nifi.connectable.Connectable;
+import org.apache.nifi.controller.ReportingTaskNode;
+
+/**
+ * Base implementation of the {@link SchedulingAgent} which encapsulates the
+ * updates to the {@link ScheduleState} based on invoked operation and then
+ * delegates to the corresponding 'do' methods. For example; By invoking
+ * {@link #schedule(Connectable, ScheduleState)} the the
+ * {@link ScheduleState#setScheduled(boolean)} with value 'true' will be
+ * invoked.
+ *
+ * @see EventDrivenSchedulingAgent
+ * @see TimerDrivenSchedulingAgent
+ * @see QuartzSchedulingAgent
+ */
+abstract class AbstractSchedulingAgent implements SchedulingAgent {
+
+ @Override
+ public void schedule(Connectable connectable, ScheduleState scheduleState) {
+ scheduleState.setScheduled(true);
+ this.doSchedule(connectable, scheduleState);
+ }
+
+ @Override
+ public void unschedule(Connectable connectable, ScheduleState scheduleState) {
+ scheduleState.setScheduled(false);
+ this.doUnschedule(connectable, scheduleState);
+ }
+
+ @Override
+ public void schedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
+ scheduleState.setScheduled(true);
+ this.doSchedule(taskNode, scheduleState);
+ }
+
+ @Override
+ public void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
+ scheduleState.setScheduled(false);
+ this.doUnschedule(taskNode, scheduleState);
+ }
+
+ /**
+ * Schedules the provided {@link Connectable}. Its {@link ScheduleState}
+ * will be set to <i>true</i>
+ *
+ * @param connectable
+ * the instance of {@link Connectable}
+ * @param scheduleState
+ * the instance of {@link ScheduleState}
+ */
+ protected abstract void doSchedule(Connectable connectable, ScheduleState scheduleState);
+
+ /**
+ * Unschedules the provided {@link Connectable}. Its {@link ScheduleState}
+ * will be set to <i>false</i>
+ *
+ * @param connectable
+ * the instance of {@link Connectable}
+ * @param scheduleState
+ * the instance of {@link ScheduleState}
+ */
+ protected abstract void doUnschedule(Connectable connectable, ScheduleState scheduleState);
+
+ /**
+ * Schedules the provided {@link ReportingTaskNode}. Its
+ * {@link ScheduleState} will be set to <i>true</i>
+ *
+ * @param connectable
+ * the instance of {@link ReportingTaskNode}
+ * @param scheduleState
+ * the instance of {@link ScheduleState}
+ */
+ protected abstract void doSchedule(ReportingTaskNode connectable, ScheduleState scheduleState);
+
+ /**
+ * Unschedules the provided {@link ReportingTaskNode}. Its
+ * {@link ScheduleState} will be set to <i>false</i>
+ *
+ * @param connectable
+ * the instance of {@link ReportingTaskNode}
+ * @param scheduleState
+ * the instance of {@link ScheduleState}
+ */
+ protected abstract void doUnschedule(ReportingTaskNode connectable, ScheduleState scheduleState);
+}
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/EventDrivenSchedulingAgent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
index 76d7c08..37cab01 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/EventDrivenSchedulingAgent.java
@@ -51,7 +51,7 @@ import org.apache.nifi.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class EventDrivenSchedulingAgent implements SchedulingAgent {
+public class EventDrivenSchedulingAgent extends AbstractSchedulingAgent {
private static final Logger logger = LoggerFactory.getLogger(EventDrivenSchedulingAgent.class);
@@ -94,24 +94,24 @@ public class EventDrivenSchedulingAgent implements SchedulingAgent {
}
@Override
- public void schedule(final ReportingTaskNode taskNode, ScheduleState scheduleState) {
+ public void doSchedule(final ReportingTaskNode taskNode, ScheduleState scheduleState) {
throw new UnsupportedOperationException("ReportingTasks cannot be scheduled in Event-Driven Mode");
}
@Override
- public void unschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
+ public void doUnschedule(ReportingTaskNode taskNode, ScheduleState scheduleState) {
throw new UnsupportedOperationException("ReportingTasks cannot be scheduled in Event-Driven Mode");
}
@Override
- public void schedule(final Connectable connectable, final ScheduleState scheduleState) {
+ public void doSchedule(final Connectable connectable, final ScheduleState scheduleState) {
workerQueue.resumeWork(connectable);
logger.info("Scheduled {} to run in Event-Driven mode", connectable);
scheduleStates.put(connectable, scheduleState);
}
@Override
- public void unschedule(final Connectable connectable, final ScheduleState scheduleState) {
+ public void doUnschedule(final Connectable connectable, final ScheduleState scheduleState) {
workerQueue.suspendWork(connectable);
logger.info("Stopped scheduling {} to run", connectable);
}
[6/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
Posted by ma...@apache.org.
NIFI-1464, Refactored Processor's life-cycle operation sequence
* Simplified and cleaned StandardProcessScheduler.start/stopProcessor methods
* Added stop/start operations to ProcessorNode.
* Removed unnecessary synchronization blocks related to ScheduledState in favor of enforcing order and idempotency via CAS operations. Those synchronization blocks were causing intermittent deadlocks whenever @OnScheduled blocks indefinitely.
* Added support for stopping the service when @OnScheduled operation hangs.
* Fixed the order of life-cycle operation invocation ensuring that each operation can *only* be invoked at the appropriate time
* Removed unnecessary locks from StandardProcessNode since Atomic variables are used.
* Removed calls to @OnStopped from ContinuallyRunningProcessTask while ensuring that procesor's full shut down in implementation of StandardProcessorNode.stop() method.
* Removed dead code
* Added comprehensive tests suite that covers 95% of Processor's life-cycle operations within the scope of FlowController, StandardProcesssScheduler and StandardProcessNode
* Improved and added javadocs on covered operations with detailed explanations.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/0c5b1c27
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/0c5b1c27
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/0c5b1c27
Branch: refs/heads/master
Commit: 0c5b1c27f26a008baedac4eb0c69f9b01080eaec
Parents: f6705f2
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Thu Feb 4 16:23:45 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Mar 11 12:54:50 2016 -0500
----------------------------------------------------------------------
.../apache/nifi/controller/ScheduledState.java | 6 +-
.../org/apache/nifi/util/NiFiProperties.java | 2 +
.../src/main/asciidoc/administration-guide.adoc | 1 +
.../controller/AbstractConfiguredComponent.java | 7 +-
.../apache/nifi/controller/ProcessorNode.java | 50 +-
.../nifi/controller/StandardProcessorNode.java | 1135 ++++++++++--------
.../scheduling/AbstractSchedulingAgent.java | 103 ++
.../scheduling/EventDrivenSchedulingAgent.java | 10 +-
.../scheduling/QuartzSchedulingAgent.java | 10 +-
.../controller/scheduling/ScheduleState.java | 8 +-
.../scheduling/StandardProcessScheduler.java | 218 +---
.../scheduling/TimerDrivenSchedulingAgent.java | 10 +-
.../StandardControllerServiceProvider.java | 1 +
.../tasks/ContinuallyRunProcessorTask.java | 12 -
.../scheduling/TestProcessorLifecycle.java | 760 ++++++++++++
.../TestStandardControllerServiceProvider.java | 19 +-
.../src/test/resources/state-management.xml | 38 +
17 files changed, 1654 insertions(+), 736 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java
----------------------------------------------------------------------
diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java b/nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java
index d482eae..bf608db 100644
--- a/nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java
+++ b/nifi-api/src/main/java/org/apache/nifi/controller/ScheduledState.java
@@ -33,5 +33,9 @@ public enum ScheduledState {
/**
* Entity is currently scheduled to run
*/
- RUNNING;
+ RUNNING,
+
+ STARTING,
+
+ STOPPING;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index c432edf..2be8177 100644
--- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -71,6 +71,7 @@ public class NiFiProperties extends Properties {
public static final String ADMINISTRATIVE_YIELD_DURATION = "nifi.administrative.yield.duration";
public static final String PERSISTENT_STATE_DIRECTORY = "nifi.persistent.state.directory";
public static final String BORED_YIELD_DURATION = "nifi.bored.yield.duration";
+ public static final String PROCESSOR_START_TIMEOUT = "nifi.processor.start.timeout";
// content repository properties
public static final String REPOSITORY_CONTENT_PREFIX = "nifi.content.repository.directory.";
@@ -539,6 +540,7 @@ public class NiFiProperties extends Properties {
return shouldSupport;
}
+ @SuppressWarnings("unchecked")
public Set<String> getAnonymousAuthorities() {
final Set<String> authorities;
http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-docs/src/main/asciidoc/administration-guide.adoc
----------------------------------------------------------------------
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index 682a92c..58fe205 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -1153,6 +1153,7 @@ nifi.nar.library.directory.lib2=/nars/lib2 +
Providing three total locations, including _nifi.nar.library.directory_.
|nifi.nar.working.directory|The location of the nar working directory. The default value is ./work/nar and probably should be left as is.
|nifi.documentation.working.directory|The documentation working directory. The default value is ./work/docs/components and probably should be left as is.
+|nifi.processor.start.timeout|Time (milliseconds) to wait for a Processors to start before other life-cycle operation (e.g., stop) could be invoked. Default is infinite.
|====
http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
index 924e61e..65e59ee 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractConfiguredComponent.java
@@ -44,7 +44,6 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone
private final ConfigurableComponent component;
private final ValidationContextFactory validationContextFactory;
private final ControllerServiceProvider serviceProvider;
-
private final AtomicReference<String> name;
private final AtomicReference<String> annotationData = new AtomicReference<>();
@@ -298,4 +297,10 @@ public abstract class AbstractConfiguredComponent implements ConfigurableCompone
public abstract void verifyModifiable() throws IllegalStateException;
+ /**
+ *
+ */
+ ControllerServiceProvider getControllerServiceProvider() {
+ return this.serviceProvider;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/0c5b1c27/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 d340c77..d7dbe24 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
@@ -18,6 +18,8 @@ package org.apache.nifi.controller;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
@@ -25,6 +27,7 @@ import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.logging.LogLevel;
+import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.scheduling.SchedulingStrategy;
@@ -54,8 +57,6 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
@Override
public abstract boolean isValid();
- public abstract void setScheduledState(ScheduledState scheduledState);
-
public abstract void setBulletinLevel(LogLevel bulletinLevel);
public abstract LogLevel getBulletinLevel();
@@ -99,4 +100,49 @@ public abstract class ProcessorNode extends AbstractConfiguredComponent implemen
*/
public abstract void verifyCanStart(Set<ControllerServiceNode> ignoredReferences);
+ /**
+ * Will start the {@link Processor} represented by this
+ * {@link ProcessorNode}. Starting processor typically means invoking it's
+ * operation that is annotated with @OnScheduled and then executing a
+ * callback provided by the {@link ProcessScheduler} to which typically
+ * initiates
+ * {@link Processor#onTrigger(ProcessContext, org.apache.nifi.processor.ProcessSessionFactory)}
+ * cycle.
+ *
+ * @param scheduler
+ * implementation of {@link ScheduledExecutorService} used to
+ * initiate processor <i>start</i> task
+ * @param administrativeYieldMillis
+ * the amount of milliseconds to wait for administrative yield
+ * @param processContext
+ * the instance of {@link ProcessContext} and
+ * {@link ControllerServiceLookup}
+ * @param schedulingAgentCallback
+ * the callback provided by the {@link ProcessScheduler} to
+ * execute upon successful start of the Processor
+ */
+ public abstract <T extends ProcessContext & ControllerServiceLookup> void start(ScheduledExecutorService scheduler,
+ long administrativeYieldMillis, T processContext, Runnable schedulingAgentCallback);
+
+ /**
+ * Will stop the {@link Processor} represented by this {@link ProcessorNode}.
+ * Stopping processor typically means invoking it's operation that is
+ * annotated with @OnUnschedule and then @OnStopped.
+ *
+ * @param scheduler
+ * implementation of {@link ScheduledExecutorService} used to
+ * initiate processor <i>stop</i> task
+ * @param processContext
+ * the instance of {@link ProcessContext} and
+ * {@link ControllerServiceLookup}
+ * @param activeThreadMonitorCallback
+ * the callback provided by the {@link ProcessScheduler} to
+ * report the count of processor's active threads. Typically it
+ * is used to ensure that operations annotated with @OnUnschedule
+ * and then @OnStopped are not invoked until such count reaches
+ * 0, essentially allowing tasks to finish before bringing
+ * processor to a halt.
+ */
+ public abstract <T extends ProcessContext & ControllerServiceLookup> void stop(ScheduledExecutorService scheduler,
+ T processContext, Callable<Boolean> activeThreadMonitorCallback);
}