Apache/nifi
1
0
Fork 0
mirror of https://github.com/apache/nifi.git synced 2025-02-18 07:56:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

NIFI-4: Fixed documentation of OnScheduled and OnUnscheduled. Updated StandardProcessScheduler to invoke OnScheduled, OnUnscheduled, OnStopped methods appropriately.

Browse Source
This commit is contained in:
Mark Payne 2015-01-16 13:35:17 -05:00
parent 6b5d1a86be
commit 68707ce3c4
3 changed files with 40 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -161,18 +161,21 @@ public final class StandardProcessScheduler implements ProcessScheduler {
scheduleState.setScheduled(true); scheduleState.setScheduled(true);
final Runnable startReportingTaskRunnable = new Runnable() { final Runnable startReportingTaskRunnable = new Runnable() {
@SuppressWarnings("deprecation")
@Override @Override
public void run() { public void run() {
// Continually attempt to start the Reporting Task, and if we fail sleep for a bit each time.
while (true) { while (true) {
final ReportingTask reportingTask = taskNode.getReportingTask(); final ReportingTask reportingTask = taskNode.getReportingTask();
try { try {
try (final NarCloseable x = NarCloseable.withNarLoader()) { try (final NarCloseable x = NarCloseable.withNarLoader()) {
ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, reportingTask, taskNode.getConfigurationContext()); ReflectionUtils.invokeMethodsWithAnnotation(OnConfigured.class, OnScheduled.class, reportingTask, taskNode.getConfigurationContext());
} }
break; break;
} catch (final InvocationTargetException ite) { } catch (final InvocationTargetException ite) {
LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}", LOG.error("Failed to invoke the On-Scheduled Lifecycle methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}",
new Object[]{reportingTask, ite.getTargetException(), administrativeYieldDuration}); new Object[]{reportingTask, ite.getTargetException(), administrativeYieldDuration});
LOG.error("", ite.getTargetException()); LOG.error("", ite.getTargetException());
@ -181,7 +184,7 @@ public final class StandardProcessScheduler implements ProcessScheduler {
} catch (final InterruptedException ie) { } catch (final InterruptedException ie) {
} }
} catch (final Exception e) { } catch (final Exception e) {
LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}", LOG.error("Failed to invoke the On-Scheduled Lifecycle methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}",
new Object[]{reportingTask, e.toString(), administrativeYieldDuration}, e); new Object[]{reportingTask, e.toString(), administrativeYieldDuration}, e);
try { try {
Thread.sleep(administrativeYieldMillis); Thread.sleep(administrativeYieldMillis);
@ -213,34 +216,31 @@ public final class StandardProcessScheduler implements ProcessScheduler {
public void run() { public void run() {
final ConfigurationContext configurationContext = taskNode.getConfigurationContext(); final ConfigurationContext configurationContext = taskNode.getConfigurationContext();
while (true) { try {
try { try (final NarCloseable x = NarCloseable.withNarLoader()) {
try (final NarCloseable x = NarCloseable.withNarLoader()) { ReflectionUtils.invokeMethodsWithAnnotation(OnUnscheduled.class, org.apache.nifi.processor.annotation.OnUnscheduled.class, reportingTask, configurationContext);
ReflectionUtils.invokeMethodsWithAnnotation(OnUnscheduled.class, org.apache.nifi.processor.annotation.OnUnscheduled.class, reportingTask, configurationContext); }
} } catch (final InvocationTargetException ite) {
break; LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}",
} catch (final InvocationTargetException ite) { new Object[]{reportingTask, ite.getTargetException(), administrativeYieldDuration});
LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}", LOG.error("", ite.getTargetException());
new Object[]{reportingTask, ite.getTargetException(), administrativeYieldDuration});
LOG.error("", ite.getTargetException());
try { try {
Thread.sleep(administrativeYieldMillis); Thread.sleep(administrativeYieldMillis);
} catch (final InterruptedException ie) { } catch (final InterruptedException ie) {
} }
} catch (final Exception e) { } catch (final Exception e) {
LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}", LOG.error("Failed to invoke the @OnConfigured methods of {} due to {}; administratively yielding this ReportingTask and will attempt to schedule it again after {}",
new Object[]{reportingTask, e.toString(), administrativeYieldDuration}, e); new Object[]{reportingTask, e.toString(), administrativeYieldDuration}, e);
try { try {
Thread.sleep(administrativeYieldMillis); Thread.sleep(administrativeYieldMillis);
} catch (final InterruptedException ie) { } catch (final InterruptedException ie) {
}
} }
} }
agent.unschedule(taskNode, scheduleState); agent.unschedule(taskNode, scheduleState);
if (scheduleState.getActiveThreadCount() == 0) { if (scheduleState.getActiveThreadCount() == 0 && scheduleState.mustCallOnStoppedMethods()) {
ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, org.apache.nifi.processor.annotation.OnStopped.class, reportingTask, configurationContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, org.apache.nifi.processor.annotation.OnStopped.class, reportingTask, configurationContext);
} }
} }

2
nifi/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnScheduled.java
View File

@ -41,7 +41,7 @@ import java.lang.annotation.Target;
* *
* <p> * <p>
* If using 1 argument and the component using the annotation is a Reporting Task, that argument must * If using 1 argument and the component using the annotation is a Reporting Task, that argument must
* be of type {@link org.apache.nifi.reporting.ReportingContext ReportingContext}. * be of type {@link org.apache.nifi.controller.ConfigurationContext ConfigurationContext}.
* </p> * </p>
* *
* If any method annotated with this annotation throws any Throwable, the framework will wait a while * If any method annotated with this annotation throws any Throwable, the framework will wait a while

14
nifi/nifi-api/src/main/java/org/apache/nifi/annotation/lifecycle/OnUnscheduled.java
View File

@ -33,6 +33,20 @@ import java.lang.annotation.Target;
* threads are potentially running. To invoke a method after all threads have * threads are potentially running. To invoke a method after all threads have
* finished processing, see the {@link OnStopped} annotation. * finished processing, see the {@link OnStopped} annotation.
* </p> * </p>
*
* <p>
* Methods using this annotation must take either 0 arguments or a single argument.
* </p>
*
* <p>
* If using 1 argument and the component using the annotation is a Processor, that argument must
* be of type {@link org.apache.nifi.processor.ProcessContext ProcessContext}.
* </p>
*
* <p>
* If using 1 argument and the component using the annotation is a Reporting Task, that argument must
* be of type {@link org.apache.nifi.controller.ConfigurationContext ConfigurationContext}.
* </p>
* *
* @author none * @author none
*/ */