This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 337069a4eb2 NIFI-16088 Add scheduling diagnostics for cron
primary-node-only processors (#11407)
337069a4eb2 is described below
commit 337069a4eb2f57fc194855df42b16d0e8586f474
Author: Kevin Doran <[email protected]>
AuthorDate: Wed Jul 8 06:40:12 2026 -0400
NIFI-16088 Add scheduling diagnostics for cron primary-node-only processors
(#11407)
Cron triggers are scheduled as one-shot tasks that reschedule only after a
successful run. If the triggered task throws, or the next trigger fails to
be
scheduled, the scheduling chain silently terminates: the component stays
RUNNING but never fires again on its schedule until it is stopped and
started,
with no record in the logs. Make that observable:
- CronSchedulingAgent: log a WARN (before propagating) when a cron-scheduled
trigger throws or when rescheduling the next trigger fails, noting the
component will not fire again until stopped and started.
- ConnectableTask: log at INFO (was DEBUG) when a cron-driven,
primary-node-only
component is skipped because this is not the primary node, so a skipped
scheduled fire is visible; timer-driven components stay at DEBUG to avoid
noise.
Diagnostics only; no scheduling or lifecycle behavior is changed.
---
.../controller/scheduling/CronSchedulingAgent.java | 23 ++++++++++++--
.../nifi/controller/tasks/ConnectableTask.java | 10 +++++-
.../scheduling/CronSchedulingAgentTest.java | 36 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 4 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/CronSchedulingAgent.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/CronSchedulingAgent.java
index f44f1b813ce..17eae44f8c7 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/CronSchedulingAgent.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/CronSchedulingAgent.java
@@ -129,8 +129,17 @@ public class CronSchedulingAgent extends
AbstractTimeBasedSchedulingAgent {
try {
continuallyRunTask.invoke();
} catch (final RuntimeException re) {
+ // Diagnostic: an exception here skips the
rescheduling below. Because each cron trigger is a
+ // one-shot scheduled task that reschedules only after
a successful run, a throwable silently
+ // terminates the scheduling chain: the component
stays RUNNING but never fires again on its
+ // schedule until it is stopped and started. Log it
before propagating, since the executor
+ // otherwise discards it with no record.
+ logger.warn("Cron-scheduled trigger for {} threw an
exception; its scheduling chain will terminate "
+ + "and it will not run again on its schedule until
it is stopped and started", connectable, re);
throw re;
} catch (final Exception e) {
+ logger.warn("Cron-scheduled trigger for {} threw an
exception; its scheduling chain will terminate "
+ + "and it will not run again on its schedule until
it is stopped and started", connectable, e);
throw new ProcessException(e);
}
@@ -138,9 +147,17 @@ public class CronSchedulingAgent extends
AbstractTimeBasedSchedulingAgent {
final long delay = getDelay(nextSchedule);
logger.debug("Finished task for {}; next scheduled time is
at {} after a delay of {} milliseconds", connectable, nextSchedule, delay);
- final ScheduledFuture<?> newFuture =
flowEngine.schedule(this, delay, TimeUnit.MILLISECONDS);
- final ScheduledFuture<?> oldFuture =
componentFuturesMap.put(taskNumber.get(), newFuture);
- scheduleState.replaceFuture(oldFuture, newFuture);
+ try {
+ final ScheduledFuture<?> newFuture =
flowEngine.schedule(this, delay, TimeUnit.MILLISECONDS);
+ final ScheduledFuture<?> oldFuture =
componentFuturesMap.put(taskNumber.get(), newFuture);
+ scheduleState.replaceFuture(oldFuture, newFuture);
+ } catch (final RuntimeException re) {
+ // Diagnostic: failing to schedule the next trigger
(e.g. the engine is shutting down) also
+ // silently terminates the scheduling chain, leaving
the component RUNNING but unscheduled.
+ logger.warn("Failed to reschedule the next cron
trigger for {}; its scheduling chain will terminate "
+ + "and it will not run again on its schedule until
it is stopped and started", connectable, re);
+ throw re;
+ }
}
};
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
index 214c1c94fa0..204da4a1cf1 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ConnectableTask.java
@@ -49,6 +49,7 @@ import org.apache.nifi.processor.SimpleProcessLogger;
import org.apache.nifi.processor.StandardProcessContext;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.exception.TerminatedTaskException;
+import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.Connectables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -177,7 +178,14 @@ public class ConnectableTask {
// make sure that either we're not clustered or this processor runs on
all nodes or that this is the primary node
if (!isRunOnCluster(flowController)) {
- logger.debug("Will not trigger {} because this is not the primary
node", connectable);
+ // Diagnostic: a cron-driven primary-node-only component triggers
comparatively rarely, so log at INFO to
+ // make a skipped scheduled fire visible (e.g. a component that
stops firing on its schedule after a
+ // primary-node change). Timer-driven components trigger
frequently, so keep those at DEBUG to avoid noise.
+ if (connectable.getSchedulingStrategy() ==
SchedulingStrategy.CRON_DRIVEN) {
+ logger.info("Will not trigger {} because this is not the
primary node; the cron-scheduled fire is being skipped", connectable);
+ } else {
+ logger.debug("Will not trigger {} because this is not the
primary node", connectable);
+ }
return InvocationResult.yield("This node is not the primary node");
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/CronSchedulingAgentTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/CronSchedulingAgentTest.java
index 3b46865ae51..10a61abd38a 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/CronSchedulingAgentTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/CronSchedulingAgentTest.java
@@ -27,14 +27,20 @@ import org.apache.nifi.reporting.ReportingTask;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -93,6 +99,36 @@ class CronSchedulingAgentTest {
schedulingAgent.doSchedule(connectable, lifecycleState);
}
+ @Test
+ void testCronTaskExceptionPropagatesRatherThanSilentlyStoppingChain() {
+ // A cron trigger is a one-shot task that reschedules only after a
successful run. If the triggered task
+ // throws, the exception must still propagate (it is also logged as a
diagnostic) rather than being silently
+ // swallowed in a way that could leave the component RUNNING but
unscheduled. This guards that propagation.
+ final String componentId = UUID.randomUUID().toString();
+ final LifecycleState lifecycleState = new LifecycleState(componentId);
+
+
when(connectable.evaluateParameters(eq(DEFAULT_CRON_EXPRESSION))).thenReturn(DEFAULT_CRON_EXPRESSION);
+
when(connectable.getSchedulingPeriod()).thenReturn(DEFAULT_CRON_EXPRESSION);
+ when(connectable.getMaxConcurrentTasks()).thenReturn(CONCURRENT_TASKS);
+ when(connectable.getIdentifier()).thenReturn(componentId);
+
when(flowController.getStateManagerProvider()).thenReturn(stateManagerProvider);
+
when(stateManagerProvider.getStateManager(eq(componentId))).thenReturn(stateManager);
+
when(flowController.getGarbageCollectionLog()).thenReturn(mock(GarbageCollectionLog.class));
+
+ schedulingAgent.doSchedule(connectable, lifecycleState);
+
+ // Capture the self-rescheduling command submitted to the engine and
run it with a triggered task that throws.
+ final ArgumentCaptor<Runnable> commandCaptor =
ArgumentCaptor.forClass(Runnable.class);
+ verify(flowEngine).schedule(commandCaptor.capture(), anyLong(),
any(TimeUnit.class));
+ final Runnable command = commandCaptor.getValue();
+
+ final RuntimeException failure = new RuntimeException("trigger
failure");
+ when(connectable.getYieldExpiration()).thenThrow(failure);
+
+ final RuntimeException thrown = assertThrows(RuntimeException.class,
command::run);
+ assertEquals(failure, thrown);
+ }
+
@Test
void testDoScheduleReportingTaskNode() {
final String componentId = UUID.randomUUID().toString();
