You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2018/07/12 01:31:13 UTC
[05/10] nifi git commit: NIFI-5361: When submitting many processors
to start,
calculate the 'timeout timestamp' immediately before calling @OnScheduled
method, after the task has been scheduled to run,
instead of before the task has a chance to run.
NIFI-5361: When submitting many processors to start, calculate the 'timeout timestamp' immediately before calling @OnScheduled method, after the task has been scheduled to run, instead of before the task has a chance to run.
This closes #2831
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/e09ab9b6
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/e09ab9b6
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/e09ab9b6
Branch: refs/heads/support/nifi-1.7.x
Commit: e09ab9b69a3bd21c09118dd2a2c9b8bd4e091bae
Parents: 260bc29
Author: Mark Payne <ma...@hotmail.com>
Authored: Fri Jun 29 09:25:57 2018 -0400
Committer: Andy LoPresto <al...@apache.org>
Committed: Wed Jul 11 18:29:31 2018 -0700
----------------------------------------------------------------------
.../org/apache/nifi/controller/StandardProcessorNode.java | 8 ++++++--
.../nifi/controller/scheduling/StandardProcessScheduler.java | 2 +-
2 files changed, 7 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/e09ab9b6/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 6f51f87..413f097 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
@@ -1481,12 +1481,16 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
final Processor processor = getProcessor();
final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor);
- final long completionTimestamp = System.currentTimeMillis() + onScheduleTimeoutMillis;
+ // Completion Timestamp is set to MAX_VALUE because we don't want to timeout until the task has a chance to run.
+ final AtomicLong completionTimestampRef = new AtomicLong(Long.MAX_VALUE);
// Create a task to invoke the @OnScheduled annotation of the processor
final Callable<Void> startupTask = () -> {
LOG.debug("Invoking @OnScheduled methods of {}", processor);
+ // Now that the task has been scheduled, set the timeout
+ completionTimestampRef.set(System.currentTimeMillis() + onScheduleTimeoutMillis);
+
try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) {
try {
activateThread();
@@ -1571,7 +1575,7 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable
return;
}
- monitorAsyncTask(taskFuture, monitoringFuture, completionTimestamp);
+ monitorAsyncTask(taskFuture, monitoringFuture, completionTimestampRef.get());
}
};
http://git-wip-us.apache.org/repos/asf/nifi/blob/e09ab9b6/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 8f7eb2f..b23e763 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
@@ -319,7 +319,7 @@ public final class StandardProcessScheduler implements ProcessScheduler {
@Override
public Future<?> scheduleTask(final Callable<?> task) {
lifecycleState.incrementActiveThreadCount(null);
- return componentMonitoringThreadPool.submit(task);
+ return componentLifeCycleThreadPool.submit(task);
}
@Override