Repository: nifi Updated Branches: refs/heads/master aca2a92d6 -> 7ba10a6de
NIFI-2484: Ensure that if a Processor throws an Exception from an @OnScheduled method that @OnStopped methods get called This closes #791. Signed-off-by: Bryan Bende <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/7ba10a6d Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/7ba10a6d Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/7ba10a6d Branch: refs/heads/master Commit: 7ba10a6deaf2b580ff02431a81dd04b8e5065a53 Parents: aca2a92 Author: Mark Payne <[email protected]> Authored: Thu Aug 4 13:05:03 2016 -0400 Committer: Bryan Bende <[email protected]> Committed: Thu Aug 4 15:34:10 2016 -0400 ---------------------------------------------------------------------- .../nifi/controller/StandardProcessorNode.java | 2 + .../scheduling/StandardProcessScheduler.java | 3 + .../controller/TestStandardProcessorNode.java | 182 +++++++++++++++++++ 3 files changed, 187 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/7ba10a6d/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 ce9ae96..ab37aee 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 @@ -1258,6 +1258,8 @@ public class StandardProcessorNode extends ProcessorNode implements Connectable LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.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 { http://git-wip-us.apache.org/repos/asf/nifi/blob/7ba10a6d/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 feabd87..aadbff0 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 @@ -219,6 +219,9 @@ public final class StandardProcessScheduler implements ProcessScheduler { + "ReportingTask and will attempt to schedule it again after {}", new Object[] { reportingTask, e.toString(), administrativeYieldDuration }, e); + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, reportingTask, taskNode.getConfigurationContext()); + ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, reportingTask, taskNode.getConfigurationContext()); + try { Thread.sleep(administrativeYieldMillis); } catch (final InterruptedException ie) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7ba10a6d/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java new file mode 100644 index 0000000..3379f79 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardProcessorNode.java @@ -0,0 +1,182 @@ +/* + * 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; + +import static org.junit.Assert.assertEquals; + +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.Callable; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; + +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.PropertyValue; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.engine.FlowEngine; +import org.apache.nifi.expression.ExpressionLanguageCompiler; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.StandardProcessContext; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.util.MockPropertyValue; +import org.junit.Assert; +import org.junit.Test; + +public class TestStandardProcessorNode { + + @Test(timeout = 10000) + public void testStart() throws InterruptedException { + System.setProperty("nifi.properties.file.path", "src/test/resources/nifi.properties"); + final ProcessorThatThrowsExceptionOnScheduled processor = new ProcessorThatThrowsExceptionOnScheduled(); + final String uuid = UUID.randomUUID().toString(); + + final StandardProcessorNode procNode = new StandardProcessorNode(processor, uuid, createValidationContextFactory(), null, null); + final ScheduledExecutorService taskScheduler = new FlowEngine(2, "TestStandardProcessorNode", true); + + final StandardProcessContext processContext = new StandardProcessContext(procNode, null, null, null, null); + final SchedulingAgentCallback schedulingAgentCallback = new SchedulingAgentCallback() { + @Override + public void postMonitor() { + } + + @Override + public Future<?> invokeMonitoringTask(final Callable<?> task) { + return taskScheduler.submit(task); + } + + @Override + public void trigger() { + Assert.fail("Should not have completed"); + } + }; + + procNode.start(taskScheduler, 20000L, processContext, schedulingAgentCallback); + + Thread.sleep(1000L); + assertEquals(1, processor.onScheduledCount); + assertEquals(1, processor.onUnscheduledCount); + assertEquals(1, processor.onStoppedCount); + } + + + private ValidationContextFactory createValidationContextFactory() { + return new ValidationContextFactory() { + @Override + public ValidationContext newValidationContext(Map<PropertyDescriptor, String> properties, String annotationData, String groupId, String componentId) { + return new ValidationContext() { + + @Override + public ControllerServiceLookup getControllerServiceLookup() { + return null; + } + + @Override + public ValidationContext getControllerServiceValidationContext(ControllerService controllerService) { + return null; + } + + @Override + public ExpressionLanguageCompiler newExpressionLanguageCompiler() { + return null; + } + + @Override + public PropertyValue getProperty(PropertyDescriptor property) { + return newPropertyValue(properties.get(property)); + } + + @Override + public PropertyValue newPropertyValue(String value) { + return new MockPropertyValue(value); + } + + @Override + public Map<PropertyDescriptor, String> getProperties() { + return Collections.unmodifiableMap(properties); + } + + @Override + public String getAnnotationData() { + return null; + } + + @Override + public boolean isValidationRequired(ControllerService service) { + return false; + } + + @Override + public boolean isExpressionLanguagePresent(String value) { + return false; + } + + @Override + public boolean isExpressionLanguageSupported(String propertyName) { + return false; + } + + @Override + public String getProcessGroupIdentifier() { + return groupId; + } + }; + } + + @Override + public ValidationContext newValidationContext(Set<String> serviceIdentifiersToNotValidate, Map<PropertyDescriptor, String> properties, String annotationData, String groupId, + String componentId) { + return newValidationContext(properties, annotationData, groupId, componentId); + } + }; + + } + + + public static class ProcessorThatThrowsExceptionOnScheduled extends AbstractProcessor { + private int onScheduledCount = 0; + private int onUnscheduledCount = 0; + private int onStoppedCount = 0; + + @Override + public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + } + + @OnScheduled + public void onScheduled() { + onScheduledCount++; + throw new ProcessException("OnScheduled called - Unit Test throws Exception intentionally"); + } + + @OnUnscheduled + public void onUnscheduled() { + onUnscheduledCount++; + } + + @OnStopped + public void onStopped() { + onStoppedCount++; + } + } +}
