markap14 commented on code in PR #11164:
URL: https://github.com/apache/nifi/pull/11164#discussion_r4020804647
##########
nifi-docs/src/main/asciidoc/administration-guide.adoc:
##########
@@ -2952,6 +2952,7 @@ This cleanup mechanism takes into account only
automatically created archived _f
|`nifi.flowservice.writedelay.interval`|When many changes are made to the
_flow.json_, this property specifies how long to wait before writing out the
changes, so as to batch the changes into a single write. The default value is
`500 ms`.
|`nifi.administrative.yield.duration`|If a component allows an unexpected
exception to escape, it is considered a bug. As a result, the framework will
pause (or administratively yield) the component for this amount of time. This
is done so that the component does not use up massive amounts of system
resources, since it is known to have problems in the existing state. The
default value is `30 secs`.
|`nifi.bored.yield.duration`|When a component has no work to do (i.e., is
"bored"), this is the amount of time it will wait before checking to see if it
has new data to work on. This way, it does not use up CPU resources by checking
for new work too often. When setting this property, be aware that it could add
extra latency for components that do not constantly have work to do, as once
they go into this "bored" state, they will wait this amount of time before
checking for more work. The default value is `10 ms`.
+|`nifi.scheduling.strategy`|Selects the scheduling engine for Timer-Driven and
Cron-Driven components. `AUTO` (the default) uses virtual threads on Java 25 or
newer and standard scheduling on older Java versions. `VIRTUAL` always uses
virtual threads. On Java 21, blocking inside synchronized component code can
also block the underlying platform thread and reduce throughput. `STANDARD`
uses a fixed platform thread pool sized by
`nifi.flowcontroller.maxTimerDrivenThreadCount`.
Review Comment:
[GPT-5.6 Sol] Good point. Updated in 35dc8b9643f to refer to the Maximum
Timer Driven Thread Count in Controller Settings rather than presenting it as
an application property.
##########
nifi-framework-api/src/main/java/org/apache/nifi/diagnostics/ThreadDumpTask.java:
##########
@@ -16,116 +16,139 @@
*/
package org.apache.nifi.diagnostics;
+import com.sun.management.HotSpotDiagnosticMXBean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
import java.lang.management.LockInfo;
import java.lang.management.ManagementFactory;
import java.lang.management.MonitorInfo;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
+/**
+ * Captures platform and virtual thread stack traces when supported by the
Java runtime.
+ */
public class ThreadDumpTask implements DiagnosticTask {
+
+ private static final Logger logger =
LoggerFactory.getLogger(ThreadDumpTask.class);
+
@Override
- public DiagnosticsDumpElement captureDump(boolean verbose) {
- final ThreadMXBean mbean = ManagementFactory.getThreadMXBean();
-
- final ThreadInfo[] infos = mbean.dumpAllThreads(true, true);
- final long[] deadlockedThreadIds = mbean.findDeadlockedThreads();
- final long[] monitorDeadlockThreadIds =
mbean.findMonitorDeadlockedThreads();
-
- final List<ThreadInfo> sortedInfos = new ArrayList<>(infos.length);
- Collections.addAll(sortedInfos, infos);
- sortedInfos.sort(new Comparator<>() {
- @Override
- public int compare(ThreadInfo o1, ThreadInfo o2) {
- return
o1.getThreadName().toLowerCase().compareTo(o2.getThreadName().toLowerCase());
+ public DiagnosticsDumpElement captureDump(final boolean verbose) {
+ final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
+ final String threadDump = captureThreadDump(threadMXBean);
+
+ final StringBuilder dumpBuilder = new StringBuilder(threadDump);
+ appendDeadlockedThreadIds(dumpBuilder, "DEADLOCK DETECTED!",
threadMXBean.findDeadlockedThreads());
+ appendDeadlockedThreadIds(dumpBuilder, "MONITOR DEADLOCK DETECTED!",
threadMXBean.findMonitorDeadlockedThreads());
Review Comment:
[GPT-5.6 Sol] Updated in 35dc8b9643f. Both headings now omit the exclamation
mark.
##########
nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/scheduling/VirtualThreadStartStopCycleIT.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.tests.system.scheduling;
+
+import org.apache.nifi.tests.system.NiFiSystemIT;
+import org.apache.nifi.toolkit.client.NiFiClientException;
+import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
+import org.apache.nifi.web.api.entity.ConnectionEntity;
+import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Verifies that repeated start and stop cycles neither leak scheduling loops
nor miss invocations.
+ */
+public class VirtualThreadStartStopCycleIT extends NiFiSystemIT {
Review Comment:
[GPT-5.6 Sol] Updated in 35dc8b9643f. The new test class and method now use
package-private visibility.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/FlowControllerTest.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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 org.apache.nifi.util.NiFiProperties;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FlowControllerTest {
Review Comment:
[GPT-5.6 Sol] Agreed. Removed the test class in 35dc8b9643f and restored the
resolver method to private visibility. The Java 21 and Java 25 system-test
profiles exercise `AUTO` in the running application.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.FlowController;
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.Triggerable;
+import org.apache.nifi.controller.tasks.ConnectableTask;
+import org.apache.nifi.controller.tasks.InvocationResult;
+import org.apache.nifi.controller.tasks.ReportingTaskWrapper;
+import org.apache.nifi.nar.NarThreadContextClassLoader;
+import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.util.FormatUtils;
+import org.apache.nifi.util.NiFiProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.support.CronExpression;
+
+import java.time.OffsetDateTime;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Scheduling agent that runs components on virtual threads. A {@link
DynamicSemaphore}
+ * limits the number of component invocations that can run concurrently.
+ */
+public class VirtualThreadSchedulingAgent implements SchedulingAgent {
+
+ private static final Logger logger =
LoggerFactory.getLogger(VirtualThreadSchedulingAgent.class);
+
+ private static final long PERMIT_POLL_INTERVAL_NANOS =
TimeUnit.SECONDS.toNanos(1L);
+
+ private final FlowController flowController;
+ private final RepositoryContextFactory contextFactory;
+ private final DynamicSemaphore globalSemaphore;
+ private final long noWorkYieldNanos;
+ private final ExecutorService executorService;
+ private final ConcurrentMap<String, SchedulingGeneration>
schedulingGenerations = new ConcurrentHashMap<>();
+ private final AtomicBoolean shutdown = new AtomicBoolean();
+ private final AtomicInteger runningThreadCount = new AtomicInteger();
+ private volatile String adminYieldDuration = "1 sec";
+ private volatile long adminYieldNanos = TimeUnit.SECONDS.toNanos(1L);
+
+ public VirtualThreadSchedulingAgent(final FlowController flowController,
final RepositoryContextFactory contextFactory,
+ final NiFiProperties nifiProperties,
final int maxThreadCount) {
+ this.flowController = flowController;
+ this.contextFactory = contextFactory;
+ this.globalSemaphore = new DynamicSemaphore(maxThreadCount);
+
+ final String boredYieldDuration =
nifiProperties.getBoredYieldDuration();
+ try {
+ noWorkYieldNanos = FormatUtils.getTimeDuration(boredYieldDuration,
TimeUnit.NANOSECONDS);
+ } catch (final IllegalArgumentException e) {
+ throw new IllegalStateException("Failed to create
VirtualThreadSchedulingAgent because the "
+ + NiFiProperties.BORED_YIELD_DURATION + " property is set
to an invalid time duration: " + boredYieldDuration, e);
+ }
+
+ final ThreadFactory threadFactory = runnable -> {
+ final Thread thread =
Thread.ofVirtual().inheritInheritableThreadLocals(false).unstarted(runnable);
+
thread.setContextClassLoader(NarThreadContextClassLoader.getInstance());
+ return thread;
+ };
+ executorService = Executors.newThreadPerTaskExecutor(threadFactory);
+ logger.info("VirtualThreadSchedulingAgent initialized with {}
permits", maxThreadCount);
+ }
+
+ @Override
+ public void shutdown() {
+ signalShutdown(true);
+ executorService.shutdownNow();
+ }
+
+ public void shutdownGracefully() {
+ signalShutdown(false);
+ executorService.shutdown();
+ }
+
+ private void signalShutdown(final boolean interrupt) {
+ shutdown.set(true);
+
+ for (final SchedulingGeneration generation :
schedulingGenerations.values()) {
+ generation.stop(interrupt);
+ }
+ }
+
+ public boolean awaitTermination(final long timeout, final TimeUnit
timeUnit) throws InterruptedException {
+ return executorService.awaitTermination(timeout, timeUnit);
+ }
+
+ public boolean isTerminated() {
+ return executorService.isTerminated();
+ }
+
+ @Override
+ public void schedule(final Connectable connectable, final LifecycleState
lifecycleState) {
+ final boolean cronDriven = connectable.getSchedulingStrategy() ==
SchedulingStrategy.CRON_DRIVEN;
+ final CronExpression cronExpression;
+ final long schedulingNanos;
+ if (cronDriven) {
+ final String cronSchedule =
connectable.evaluateParameters(connectable.getSchedulingPeriod());
+ cronExpression = parseCronExpression(cronSchedule, connectable);
+ schedulingNanos = 0L;
+ } else {
+ cronExpression = null;
+ schedulingNanos =
connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS);
+ }
+
+ final String componentId = connectable.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final ConnectableTask connectableTask = new ConnectableTask(this,
connectable, flowController, contextFactory, lifecycleState);
+ final int taskCount = connectable.getMaxConcurrentTasks();
+
+ for (int i = 0; i < taskCount; i++) {
+ final String threadName = buildThreadName(connectable, i);
+ submitTask(threadName, generation, () ->
runSchedulingLoop(connectable, connectableTask, schedulingNanos,
lifecycleState, generation, cronExpression));
+ }
+
+ logger.info("Scheduled {} to run with {} virtual threads",
connectable, taskCount);
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void scheduleOnce(final Connectable connectable, final
LifecycleState lifecycleState, final Callable<Future<Void>> stopCallback) {
+ final String componentId = connectable.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final ConnectableTask connectableTask = new ConnectableTask(this,
connectable, flowController, contextFactory, lifecycleState);
+ final String threadName = buildThreadName(connectable, 0);
+
+ submitTask(threadName, generation, () -> {
+ try {
+ runOnce(connectable, connectableTask, stopCallback,
lifecycleState, generation);
+ } finally {
+ stopSchedulingGeneration(componentId, generation, false);
+ }
+ });
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void unschedule(final Connectable connectable, final LifecycleState
lifecycleState) {
+ synchronized (lifecycleState) {
+ final SchedulingGeneration generation =
schedulingGenerations.remove(connectable.getIdentifier());
+ if (generation != null) {
+ generation.stop(false);
+ }
+
+ lifecycleState.setScheduled(false);
+ }
+
+ logger.info("Stopped scheduling {} to run", connectable);
+ }
+
+ @Override
+ public void schedule(final ReportingTaskNode taskNode, final
LifecycleState lifecycleState) {
+ final boolean cronDriven = taskNode.getSchedulingStrategy() ==
SchedulingStrategy.CRON_DRIVEN;
+ final CronExpression cronExpression;
+ final long schedulingNanos;
+ if (cronDriven) {
+ cronExpression =
parseCronExpression(taskNode.getSchedulingPeriod(), taskNode);
+ schedulingNanos = 0L;
+ } else {
+ cronExpression = null;
+ schedulingNanos =
taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS);
+ }
+
+ final String componentId = taskNode.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final Runnable reportingTaskWrapper = new
ReportingTaskWrapper(taskNode, lifecycleState,
flowController.getExtensionManager());
+ final String threadName = "Reporting Task: " + taskNode.getName();
+
+ submitTask(threadName, generation,
+ () -> runReportingTaskLoop(taskNode, reportingTaskWrapper,
schedulingNanos, cronExpression, lifecycleState, generation));
+
+ logger.info("{} started on virtual thread",
taskNode.getReportingTask());
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void unschedule(final ReportingTaskNode taskNode, final
LifecycleState lifecycleState) {
+ synchronized (lifecycleState) {
+ final SchedulingGeneration generation =
schedulingGenerations.remove(taskNode.getIdentifier());
+ if (generation != null) {
+ generation.stop(false);
+ }
+
+ lifecycleState.setScheduled(false);
+ }
+
+ logger.info("Stopped scheduling {} to run",
taskNode.getReportingTask());
+ }
+
+ private SchedulingGeneration registerSchedulingGeneration(final String
componentId) {
+ if (shutdown.get()) {
+ throw new IllegalStateException("VirtualThreadSchedulingAgent has
been shut down and cannot accept new work");
+ }
+
+ final SchedulingGeneration generation = new SchedulingGeneration();
+ final SchedulingGeneration existingGeneration =
schedulingGenerations.putIfAbsent(componentId, generation);
+ if (existingGeneration != null) {
+ throw new IllegalStateException("Component " + componentId + " is
already scheduled");
+ }
+
+ if (shutdown.get()) {
+ stopSchedulingGeneration(componentId, generation, true);
+ throw new IllegalStateException("VirtualThreadSchedulingAgent has
been shut down and cannot accept new work");
+ }
+
+ return generation;
+ }
+
+ private boolean stopSchedulingGeneration(final String componentId, final
SchedulingGeneration generation, final boolean interrupt) {
+ final boolean removed = schedulingGenerations.remove(componentId,
generation);
+ generation.stop(interrupt);
+ return removed;
+ }
+
+ private boolean isActive(final LifecycleState lifecycleState, final
SchedulingGeneration generation) {
+ return !shutdown.get() && lifecycleState.isScheduled() &&
!generation.isStopped();
+ }
+
+ private static CronExpression parseCronExpression(final String
cronSchedule, final Object component) {
+ try {
+ return CronExpression.parse(cronSchedule);
+ } catch (final RuntimeException e) {
+ throw new IllegalStateException("Cannot schedule " + component + "
to run because its scheduling period is not a valid CRON expression: " +
cronSchedule, e);
+ }
+ }
+
+ @Override
+ public void onEvent(final Connectable connectable) {
+ }
+
+ @Override
+ public synchronized void setMaxThreadCount(final int maxThreads) {
+ globalSemaphore.setMaxPermits(maxThreads);
+ logger.info("Global semaphore permits updated to {}", maxThreads);
+ }
+
+ @Override
+ public synchronized void incrementMaxThreadCount(final int toAdd) {
+ if (toAdd == 0) {
+ return;
+ }
+
+ final int currentMax = globalSemaphore.getMaxPermits();
+ final int newMax = currentMax + toAdd;
+ if (newMax < 1) {
+ throw new IllegalStateException("Cannot remove " + (-toAdd) + "
permits from global semaphore because there are only " + currentMax + " permits
available");
+ }
+
+ globalSemaphore.setMaxPermits(newMax);
+ }
+
+ @Override
+ public void setAdministrativeYieldDuration(final String duration) {
+ this.adminYieldNanos = FormatUtils.getTimeDuration(duration,
TimeUnit.NANOSECONDS);
+ this.adminYieldDuration = duration;
+ }
+
+ @Override
+ public String getAdministrativeYieldDuration() {
+ return adminYieldDuration;
+ }
+
+ @Override
+ public long getAdministrativeYieldDuration(final TimeUnit timeUnit) {
+ return timeUnit.convert(adminYieldNanos, TimeUnit.NANOSECONDS);
+ }
+
+ DynamicSemaphore getGlobalSemaphore() {
+ return globalSemaphore;
+ }
+
+ int getRunningThreadCount() {
+ return runningThreadCount.get();
+ }
+
+ boolean isShutdown() {
+ return shutdown.get();
+ }
+
+ /**
+ * @return number of component invocations currently holding global permits
+ */
+ public int getActiveThreadCount() {
+ return globalSemaphore.getInUsePermits();
+ }
+
+ private void runSchedulingLoop(final Connectable connectable, final
ConnectableTask connectableTask, final long schedulingNanos,
+ final LifecycleState lifecycleState, final
SchedulingGeneration generation, final CronExpression cronExpression) {
+ final boolean cronDriven = cronExpression != null;
+
+ OffsetDateTime nextCronSchedule = null;
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(OffsetDateTime.now(),
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no future firings;
scheduling loop will exit without invoking the component", connectable);
+ return;
+ }
+
+ final long initialDelayMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ if (initialDelayMillis > 0L) {
+
waitForDelay(TimeUnit.MILLISECONDS.toNanos(initialDelayMillis), generation);
+ }
+ }
+
+ while (true) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ return;
+ }
+
+ final InvocationResult invocationResult;
+ try {
+ invocationResult = connectableTask.invoke();
+ } finally {
+ // Interrupt status from one invocation must not carry
into the scheduling loop.
+ Thread.interrupted();
+ globalSemaphore.release();
+ }
+
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(nextCronSchedule,
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no further
firings after the current invocation; scheduling loop is exiting", connectable);
+ return;
+ }
+
+ final long sleepMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ waitForDelay(TimeUnit.MILLISECONDS.toNanos(sleepMillis),
generation);
+ } else {
+ waitForNextInvocation(connectable, schedulingNanos,
generation, invocationResult);
+ }
+ } catch (final Throwable t) {
+ if (!isActive(lifecycleState, generation)) {
+ return;
+ }
+
+ try {
+ connectable.yield(adminYieldNanos, TimeUnit.NANOSECONDS);
+ } catch (final Throwable yieldError) {
+ t.addSuppressed(yieldError);
+ }
+
+ logger.error("Unexpected error in scheduling loop for {}. Will
yield for {} and continue.", connectable, adminYieldDuration, t);
+ waitForDelay(adminYieldNanos, generation);
+ }
+ }
+ }
+
+ private void runOnce(final Connectable connectable, final ConnectableTask
connectableTask, final Callable<Future<Void>> stopCallback,
+ final LifecycleState lifecycleState, final
SchedulingGeneration generation) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ if (isActive(lifecycleState, generation)) {
+ logger.warn("Run once request for {} was not executed
because permit acquisition was interrupted", connectable);
+ } else {
+ logger.warn("Run once request for {} was not executed
because scheduling is no longer active", connectable);
+ }
+
+ return;
+ }
+
+ try {
+ connectableTask.invoke();
+ } finally {
+ globalSemaphore.release();
+ }
+ } catch (final Throwable t) {
+ logger.error("Unexpected error running {} once", connectable, t);
+ } finally {
+ try {
+ stopCallback.call();
+ } catch (final Throwable t) {
+ logger.error("Error while stopping {} after running once",
connectable, t);
+ }
+ }
+ }
+
+ private void runReportingTaskLoop(final ReportingTaskNode taskNode, final
Runnable reportingTaskWrapper, final long schedulingNanos,
+ final CronExpression cronExpression,
final LifecycleState lifecycleState, final SchedulingGeneration generation) {
+ final boolean cronDriven = cronExpression != null;
+
+ OffsetDateTime nextCronSchedule = null;
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(OffsetDateTime.now(),
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no future firings;
scheduling loop will exit without invoking the reporting task",
+ taskNode.getReportingTask());
+ return;
+ }
+
+ final long initialDelayMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ if (initialDelayMillis > 0L) {
+
waitForDelay(TimeUnit.MILLISECONDS.toNanos(initialDelayMillis), generation);
+ }
+ }
+
+ while (true) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ return;
+ }
+
+ try {
+ reportingTaskWrapper.run();
+ } finally {
+ // Interrupt status from one invocation must not carry
into the scheduling loop.
+ Thread.interrupted();
+ globalSemaphore.release();
+ }
+
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(nextCronSchedule,
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no further
firings after the current invocation; scheduling loop is exiting",
+ taskNode.getReportingTask());
+ return;
+ }
+
+ final long sleepMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ waitForDelay(TimeUnit.MILLISECONDS.toNanos(sleepMillis),
generation);
+ } else {
+ waitForDelay(schedulingNanos, generation);
+ }
+ } catch (final Throwable t) {
+ if (!isActive(lifecycleState, generation)) {
+ return;
+ }
+
+ logger.error("Unexpected error in scheduling loop for {}. Will
wait for {} and continue.", taskNode.getReportingTask(), adminYieldDuration, t);
+ waitForDelay(adminYieldNanos, generation);
+ }
+ }
+ }
+
+ private void waitForNextInvocation(final Connectable connectable, final
long schedulingNanos, final SchedulingGeneration generation,
+ final InvocationResult
invocationResult) {
+ final long sleepNanos;
+ final long yieldExpiration = connectable.getYieldExpiration();
+ final long yieldDelayNanos;
+ if (yieldExpiration == 0L) {
+ yieldDelayNanos = 0L;
+ } else {
+ yieldDelayNanos =
TimeUnit.MILLISECONDS.toNanos(Math.max(yieldExpiration -
System.currentTimeMillis(), 0L));
+ }
+
+ if (yieldDelayNanos > 0L) {
+ sleepNanos = Math.max(schedulingNanos, yieldDelayNanos);
+ } else if (invocationResult.isYield()) {
+ sleepNanos = noWorkYieldNanos > 0L ? noWorkYieldNanos :
schedulingNanos;
+ } else {
+ sleepNanos = schedulingNanos;
+ }
+
+ waitForDelay(sleepNanos, generation);
+ }
+
+ private boolean acquirePermitWithPolling(final LifecycleState
lifecycleState, final SchedulingGeneration generation) {
Review Comment:
[GPT-5.6 Sol] I walked through the Stop and Run Once interleavings and this
is safe. Registration and explicit unscheduling are serialized on
`LifecycleState`. Cleanup uses `ConcurrentMap.remove(componentId, generation)`,
so completion of an older Run Once cannot remove a newer registered generation.
It then stops only the captured generation, and `SchedulingGeneration.stop()`
is idempotent. No synchronization change is needed here.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgentTest.java:
##########
@@ -0,0 +1,821 @@
+/*
+ * 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.components.state.StateManager;
+import org.apache.nifi.components.state.StateManagerProvider;
+import org.apache.nifi.connectable.Connectable;
+import org.apache.nifi.controller.FlowController;
+import org.apache.nifi.controller.GarbageCollectionLog;
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.ScheduledState;
+import org.apache.nifi.controller.repository.FlowFileEventRepository;
+import org.apache.nifi.controller.repository.RepositoryContext;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.nar.NarThreadContextClassLoader;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.reporting.ReportingContext;
+import org.apache.nifi.reporting.ReportingTask;
+import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.util.NiFiProperties;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+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 static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class VirtualThreadSchedulingAgentTest {
+
+ private static final int MAX_THREADS = 10;
+ private static final String COMPONENT_ID = UUID.randomUUID().toString();
+
+ @Mock
+ private FlowController flowController;
+
+ @Mock
+ private RepositoryContextFactory contextFactory;
+
+ @Mock
+ private NiFiProperties nifiProperties;
+
+ @Mock
+ private StateManagerProvider stateManagerProvider;
+
+ @Mock
+ private StateManager stateManager;
+
+ @Mock
+ private GarbageCollectionLog garbageCollectionLog;
+
+ @Mock
+ private ExtensionManager extensionManager;
+
+ private VirtualThreadSchedulingAgent agent;
+
+ @BeforeEach
+ void setUp() {
+ when(nifiProperties.getBoredYieldDuration()).thenReturn("10 millis");
+ agent = new VirtualThreadSchedulingAgent(flowController,
contextFactory, nifiProperties, MAX_THREADS);
+ }
+
+ @AfterEach
+ void tearDown() throws InterruptedException {
+ agent.shutdown();
+ assertTrue(agent.awaitTermination(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void testIncrementMaxThreadCountAdjustsSemaphore() {
+ final int originalPermits = agent.getGlobalSemaphore().getMaxPermits();
+
+ agent.incrementMaxThreadCount(0);
+ assertEquals(originalPermits,
agent.getGlobalSemaphore().getMaxPermits());
+
+ agent.incrementMaxThreadCount(5);
+ assertEquals(originalPermits + 5,
agent.getGlobalSemaphore().getMaxPermits());
+
+ agent.incrementMaxThreadCount(-3);
+ assertEquals(originalPermits + 2,
agent.getGlobalSemaphore().getMaxPermits());
+
+ assertThrows(IllegalStateException.class, () ->
agent.incrementMaxThreadCount(-1000));
+ }
+
+ @Test
+ void testScheduleSpawnsThreadsThatInvoke() throws InterruptedException {
+ final int concurrentTasks = 3;
+ final AtomicInteger invocationCount = new AtomicInteger(0);
+ final CountDownLatch allTasksInvoked = new
CountDownLatch(concurrentTasks);
+ final AtomicBoolean virtualThreadsUsed = new AtomicBoolean(true);
+
+ final Connectable connectable =
createMockedConnectable(concurrentTasks, SchedulingStrategy.TIMER_DRIVEN,
invocationCount, allTasksInvoked);
+ doAnswer(invocation -> {
+ invocationCount.incrementAndGet();
+ allTasksInvoked.countDown();
+ if (!Thread.currentThread().isVirtual()) {
+ virtualThreadsUsed.set(false);
+ }
+ return null;
+ }).when(connectable).onTrigger(any(), any());
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+
+ scheduleConnectable(connectable, lifecycleState);
+
+ assertTrue(allTasksInvoked.await(5, TimeUnit.SECONDS),
+ "Expected " + concurrentTasks + " threads to invoke, but only
" + (concurrentTasks - allTasksInvoked.getCount()) + " did");
+ assertTrue(invocationCount.get() >= concurrentTasks,
+ "Expected at least " + concurrentTasks + " invocations but got
" + invocationCount.get());
+ assertTrue(virtualThreadsUsed.get());
+
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void testProcessorContinuesAfterInterruptStatusSet() throws
InterruptedException {
+ final AtomicInteger invocationCount = new AtomicInteger();
+ final CountDownLatch secondInvocation = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+ doAnswer(invocation -> {
+ if (invocationCount.incrementAndGet() == 1) {
+ Thread.currentThread().interrupt();
+ } else {
+ secondInvocation.countDown();
+ }
+
+ return null;
+ }).when(connectable).onTrigger(any(), any());
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(secondInvocation.await(2, TimeUnit.SECONDS));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ }
+ }
+
+ @Test
+ void
testSchedulingThreadUsesFrameworkClassLoaderWithoutInheritedThreadLocals()
throws InterruptedException {
+ final InheritableThreadLocal<String> inheritedValue = new
InheritableThreadLocal<>();
+ final ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ final ClassLoader lifecycleClassLoader = new
ClassLoader(originalClassLoader) {
+ };
+ final AtomicReference<ClassLoader> observedClassLoader = new
AtomicReference<>();
+ final AtomicReference<String> observedInheritedValue = new
AtomicReference<>();
+ final CountDownLatch schedulingThreadObserved = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+ when(connectable.getYieldExpiration()).thenAnswer(invocation -> {
+ observedClassLoader.compareAndSet(null,
Thread.currentThread().getContextClassLoader());
+ observedInheritedValue.compareAndSet(null, inheritedValue.get());
+ schedulingThreadObserved.countDown();
+ return 0L;
+ });
+
+ inheritedValue.set("lifecycle-thread-value");
+ Thread.currentThread().setContextClassLoader(lifecycleClassLoader);
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+
+ try {
+ scheduleConnectable(connectable, lifecycleState);
+ assertTrue(schedulingThreadObserved.await(2, TimeUnit.SECONDS));
+ assertEquals(NarThreadContextClassLoader.getInstance(),
observedClassLoader.get());
+ assertNull(observedInheritedValue.get());
+ } finally {
+ inheritedValue.remove();
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ unscheduleConnectable(connectable, lifecycleState);
+ }
+ }
+
+ @Test
+ void testDuplicateScheduleIsRejected() throws InterruptedException {
+ final CountDownLatch invocationStarted = new CountDownLatch(1);
+ final CountDownLatch releaseInvocation = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+ doAnswer(invocation -> {
+ invocationStarted.countDown();
+ releaseInvocation.await();
+ return null;
+ }).when(connectable).onTrigger(any(), any());
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(invocationStarted.await(2, TimeUnit.SECONDS));
+ assertThrows(IllegalStateException.class, () ->
agent.schedule(connectable, lifecycleState));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ releaseInvocation.countDown();
+ }
+ }
+
+ @Test
+ void testComponentYieldDoesNotShortenSchedulingPeriod() throws
InterruptedException {
+ final long schedulingPeriodMillis = 500L;
+ final AtomicLong yieldExpiration = new AtomicLong();
+ final AtomicInteger invocationCount = new AtomicInteger();
+ final CountDownLatch firstInvocation = new CountDownLatch(1);
+ final CountDownLatch secondInvocation = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+
when(connectable.getSchedulingPeriod(TimeUnit.MILLISECONDS)).thenReturn(schedulingPeriodMillis);
+
when(connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(schedulingPeriodMillis));
+ when(connectable.getYieldExpiration()).thenAnswer(invocation ->
yieldExpiration.get());
+ doAnswer(invocation -> {
+ final int currentInvocation = invocationCount.incrementAndGet();
+ if (currentInvocation == 1) {
+ yieldExpiration.set(System.currentTimeMillis() + 50L);
+ firstInvocation.countDown();
+ } else if (currentInvocation == 2) {
+ secondInvocation.countDown();
+ }
+ return null;
+ }).when(connectable).onTrigger(any(), any());
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(firstInvocation.await(2, TimeUnit.SECONDS));
+ assertFalse(secondInvocation.await(250, TimeUnit.MILLISECONDS));
+ assertTrue(secondInvocation.await(2, TimeUnit.SECONDS));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ }
+ }
+
+ @Test
+ void testZeroBoredYieldUsesSchedulingPeriod() throws InterruptedException {
+ agent.shutdown();
+ when(nifiProperties.getBoredYieldDuration()).thenReturn("0 millis");
+ agent = new VirtualThreadSchedulingAgent(flowController,
contextFactory, nifiProperties, MAX_THREADS);
+
+ final long schedulingPeriodMillis = 500L;
+ final AtomicInteger schedulingAttempts = new AtomicInteger();
+ final CountDownLatch firstAttempt = new CountDownLatch(1);
+ final CountDownLatch secondAttempt = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+
when(connectable.getSchedulingPeriod(TimeUnit.MILLISECONDS)).thenReturn(schedulingPeriodMillis);
+
when(connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(schedulingPeriodMillis));
+ when(connectable.isIsolated()).thenAnswer(invocation -> {
+ final int attempt = schedulingAttempts.incrementAndGet();
+ if (attempt == 1) {
+ firstAttempt.countDown();
+ } else if (attempt == 2) {
+ secondAttempt.countDown();
+ }
+ return true;
+ });
+ when(flowController.isConfiguredForClustering()).thenReturn(true);
+ when(flowController.isPrimary()).thenReturn(false);
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(firstAttempt.await(2, TimeUnit.SECONDS));
+ assertFalse(secondAttempt.await(250, TimeUnit.MILLISECONDS));
+ assertTrue(secondAttempt.await(2, TimeUnit.SECONDS));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ }
+ }
+
+ @Test
+ void testUnscheduleWakesLongSchedulingDelay() throws InterruptedException {
+ final CountDownLatch invocationCompleted = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), invocationCompleted);
+
when(connectable.getSchedulingPeriod(TimeUnit.MILLISECONDS)).thenReturn(TimeUnit.DAYS.toMillis(1L));
+
when(connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.DAYS.toNanos(1L));
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+
+ scheduleConnectable(connectable, lifecycleState);
+ assertTrue(invocationCompleted.await(2, TimeUnit.SECONDS));
+
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+ assertEquals(0, lifecycleState.getActiveThreadCount());
+ }
+
+ @Test
+ void testScheduleOnceInvokesAndStops() throws InterruptedException {
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ final CountDownLatch stopCallbackInvoked = new CountDownLatch(1);
+
+ lifecycleState.setScheduled(true);
+ agent.scheduleOnce(connectable, lifecycleState, () -> {
+ stopCallbackInvoked.countDown();
+ return null;
+ });
+
+ assertTrue(stopCallbackInvoked.await(5, TimeUnit.SECONDS),
+ "Stop callback should have been invoked after scheduleOnce");
+ }
+
+ @Test
+ void testUnscheduleExitsWhenSemaphoreFullyContended() throws
InterruptedException {
+ agent.setMaxThreadCount(1);
+
+ final CountDownLatch releaseHeldPermit = new CountDownLatch(1);
+ final CountDownLatch permitAcquired = new CountDownLatch(1);
+ final Thread permitHolder = Thread.ofVirtual().start(() -> {
+ try {
+ agent.getGlobalSemaphore().acquire();
+ permitAcquired.countDown();
+ releaseHeldPermit.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ agent.getGlobalSemaphore().release();
+ }
+ });
+ assertTrue(permitAcquired.await(2, TimeUnit.SECONDS), "Failed to
acquire permit for test setup");
+
+ final AtomicInteger invocationCount = new AtomicInteger(0);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, invocationCount, new CountDownLatch(0));
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ waitForRunningThreadCount(1, 2, TimeUnit.SECONDS);
+ assertEquals(0, invocationCount.get());
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+
+ releaseHeldPermit.countDown();
+ permitHolder.join(1_000L);
+
+ assertEquals(0, invocationCount.get());
+ }
+
+ @Test
+ void testConcurrentIncrementMaxThreadCountIsThreadSafe() throws
InterruptedException {
+ agent.setMaxThreadCount(100);
+
+ final int threadCount = 20;
+ final int incrementsPerThread = 50;
+ final CountDownLatch start = new CountDownLatch(1);
+ final CountDownLatch done = new CountDownLatch(threadCount);
+
+ for (int i = 0; i < threadCount; i++) {
+ Thread.ofVirtual().start(() -> {
+ try {
+ start.await();
+ for (int j = 0; j < incrementsPerThread; j++) {
+ agent.incrementMaxThreadCount(1);
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+
+ start.countDown();
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+
+ assertEquals(100 + threadCount * incrementsPerThread,
agent.getGlobalSemaphore().getMaxPermits(),
+ "Lost increments imply a race condition in
incrementMaxThreadCount");
+ }
+
+ @Test
+ void testSchedulingPeriodReadOnceWhenScheduled() throws
InterruptedException {
+ final CountDownLatch invocationsCompleted = new CountDownLatch(5);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), invocationsCompleted);
+ final AtomicInteger schedulingPeriodCalls = new AtomicInteger();
+
when(connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenAnswer(invocation
-> {
+ schedulingPeriodCalls.incrementAndGet();
+ return TimeUnit.MILLISECONDS.toNanos(10L);
+ });
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(invocationsCompleted.await(5, TimeUnit.SECONDS));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+ }
+
+ assertEquals(1, schedulingPeriodCalls.get());
+ }
+
+ @Test
+ void testSchedulingLoopContinuesAfterUnexpectedError() throws
InterruptedException {
+ agent.setAdministrativeYieldDuration("1 millis");
+ final CountDownLatch invocationCompleted = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), invocationCompleted);
+ when(connectable.isIsolated()).thenThrow(new AssertionError("Simulated
scheduling error")).thenReturn(false);
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ try {
+ assertTrue(invocationCompleted.await(2, TimeUnit.SECONDS));
+ } finally {
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+ }
+
+ assertEquals(MAX_THREADS,
agent.getGlobalSemaphore().availablePermits());
+ }
+
+ @Test
+ void testInvocationExceptionStillReleasesPermit() throws
InterruptedException {
+ agent.setMaxThreadCount(2);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0));
+ final AtomicInteger invocationCount = new AtomicInteger(0);
+ final CountDownLatch successfulInvocation = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ final int count = invocationCount.incrementAndGet();
+ if (count <= 3) {
+ throw new IllegalStateException("Simulated failure " + count);
+ }
+ successfulInvocation.countDown();
+ return null;
+ }).when(connectable).onTrigger(any(), any());
+
+ final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID);
+ scheduleConnectable(connectable, lifecycleState);
+
+ assertTrue(successfulInvocation.await(2, TimeUnit.SECONDS));
+ unscheduleConnectable(connectable, lifecycleState);
+ waitForRunningThreadCount(0, 2, TimeUnit.SECONDS);
+ assertEquals(2, agent.getGlobalSemaphore().availablePermits());
+ }
+
+ @Test
+ void testCronScheduleSpawnsThreadsAndInvokes() throws InterruptedException
{
+ final AtomicInteger invocationCount = new AtomicInteger(0);
+ final CountDownLatch atLeastOneInvocation = new CountDownLatch(1);
+ final Connectable connectable = createMockedConnectable(1,
SchedulingStrategy.CRON_DRIVEN, invocationCount, atLeastOneInvocation);
+ when(connectable.getSchedulingPeriod()).thenReturn("* * * * * ?");
+ when(connectable.evaluateParameters(eq("* * * * * ?"))).thenReturn("*
* * * * ?");
Review Comment:
[GPT-5.6 Sol] Updated in 35dc8b9643f. The repeated expression is now the
shared `CRON_SCHEDULE` constant.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/ReportingTaskWrapperTest.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.tasks;
+
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.scheduling.LifecycleState;
+import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.reporting.ReportingContext;
+import org.apache.nifi.reporting.ReportingTask;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ReportingTaskWrapperTest {
Review Comment:
[GPT-5.6 Sol] This test is intentional regression coverage.
`ReportingTaskWrapper.run()` previously held its monitor while calling
`ReportingTask.onTrigger()`, which would pin the carrier thread when explicit
virtual scheduling is used on Java 21 and affect every Reporting Task. The test
verifies that framework code does not hold that monitor while extension code
runs, so I retained it.
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.FlowController;
+import org.apache.nifi.controller.ReportingTaskNode;
+import org.apache.nifi.controller.Triggerable;
+import org.apache.nifi.controller.tasks.ConnectableTask;
+import org.apache.nifi.controller.tasks.InvocationResult;
+import org.apache.nifi.controller.tasks.ReportingTaskWrapper;
+import org.apache.nifi.nar.NarThreadContextClassLoader;
+import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.util.FormatUtils;
+import org.apache.nifi.util.NiFiProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.support.CronExpression;
+
+import java.time.OffsetDateTime;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Scheduling agent that runs components on virtual threads. A {@link
DynamicSemaphore}
+ * limits the number of component invocations that can run concurrently.
+ */
+public class VirtualThreadSchedulingAgent implements SchedulingAgent {
+
+ private static final Logger logger =
LoggerFactory.getLogger(VirtualThreadSchedulingAgent.class);
+
+ private static final long PERMIT_POLL_INTERVAL_NANOS =
TimeUnit.SECONDS.toNanos(1L);
+
+ private final FlowController flowController;
+ private final RepositoryContextFactory contextFactory;
+ private final DynamicSemaphore globalSemaphore;
+ private final long noWorkYieldNanos;
+ private final ExecutorService executorService;
+ private final ConcurrentMap<String, SchedulingGeneration>
schedulingGenerations = new ConcurrentHashMap<>();
+ private final AtomicBoolean shutdown = new AtomicBoolean();
+ private final AtomicInteger runningThreadCount = new AtomicInteger();
+ private volatile String adminYieldDuration = "1 sec";
+ private volatile long adminYieldNanos = TimeUnit.SECONDS.toNanos(1L);
+
+ public VirtualThreadSchedulingAgent(final FlowController flowController,
final RepositoryContextFactory contextFactory,
+ final NiFiProperties nifiProperties,
final int maxThreadCount) {
+ this.flowController = flowController;
+ this.contextFactory = contextFactory;
+ this.globalSemaphore = new DynamicSemaphore(maxThreadCount);
+
+ final String boredYieldDuration =
nifiProperties.getBoredYieldDuration();
+ try {
+ noWorkYieldNanos = FormatUtils.getTimeDuration(boredYieldDuration,
TimeUnit.NANOSECONDS);
+ } catch (final IllegalArgumentException e) {
+ throw new IllegalStateException("Failed to create
VirtualThreadSchedulingAgent because the "
+ + NiFiProperties.BORED_YIELD_DURATION + " property is set
to an invalid time duration: " + boredYieldDuration, e);
+ }
+
+ final ThreadFactory threadFactory = runnable -> {
+ final Thread thread =
Thread.ofVirtual().inheritInheritableThreadLocals(false).unstarted(runnable);
+
thread.setContextClassLoader(NarThreadContextClassLoader.getInstance());
+ return thread;
+ };
+ executorService = Executors.newThreadPerTaskExecutor(threadFactory);
+ logger.info("VirtualThreadSchedulingAgent initialized with {}
permits", maxThreadCount);
+ }
+
+ @Override
+ public void shutdown() {
+ signalShutdown(true);
+ executorService.shutdownNow();
+ }
+
+ public void shutdownGracefully() {
+ signalShutdown(false);
+ executorService.shutdown();
+ }
+
+ private void signalShutdown(final boolean interrupt) {
+ shutdown.set(true);
+
+ for (final SchedulingGeneration generation :
schedulingGenerations.values()) {
+ generation.stop(interrupt);
+ }
+ }
+
+ public boolean awaitTermination(final long timeout, final TimeUnit
timeUnit) throws InterruptedException {
+ return executorService.awaitTermination(timeout, timeUnit);
+ }
+
+ public boolean isTerminated() {
+ return executorService.isTerminated();
+ }
+
+ @Override
+ public void schedule(final Connectable connectable, final LifecycleState
lifecycleState) {
+ final boolean cronDriven = connectable.getSchedulingStrategy() ==
SchedulingStrategy.CRON_DRIVEN;
+ final CronExpression cronExpression;
+ final long schedulingNanos;
+ if (cronDriven) {
+ final String cronSchedule =
connectable.evaluateParameters(connectable.getSchedulingPeriod());
+ cronExpression = parseCronExpression(cronSchedule, connectable);
+ schedulingNanos = 0L;
+ } else {
+ cronExpression = null;
+ schedulingNanos =
connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS);
+ }
+
+ final String componentId = connectable.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final ConnectableTask connectableTask = new ConnectableTask(this,
connectable, flowController, contextFactory, lifecycleState);
+ final int taskCount = connectable.getMaxConcurrentTasks();
+
+ for (int i = 0; i < taskCount; i++) {
+ final String threadName = buildThreadName(connectable, i);
+ submitTask(threadName, generation, () ->
runSchedulingLoop(connectable, connectableTask, schedulingNanos,
lifecycleState, generation, cronExpression));
+ }
+
+ logger.info("Scheduled {} to run with {} virtual threads",
connectable, taskCount);
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void scheduleOnce(final Connectable connectable, final
LifecycleState lifecycleState, final Callable<Future<Void>> stopCallback) {
+ final String componentId = connectable.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final ConnectableTask connectableTask = new ConnectableTask(this,
connectable, flowController, contextFactory, lifecycleState);
+ final String threadName = buildThreadName(connectable, 0);
+
+ submitTask(threadName, generation, () -> {
+ try {
+ runOnce(connectable, connectableTask, stopCallback,
lifecycleState, generation);
+ } finally {
+ stopSchedulingGeneration(componentId, generation, false);
+ }
+ });
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void unschedule(final Connectable connectable, final LifecycleState
lifecycleState) {
+ synchronized (lifecycleState) {
+ final SchedulingGeneration generation =
schedulingGenerations.remove(connectable.getIdentifier());
+ if (generation != null) {
+ generation.stop(false);
+ }
+
+ lifecycleState.setScheduled(false);
+ }
+
+ logger.info("Stopped scheduling {} to run", connectable);
+ }
+
+ @Override
+ public void schedule(final ReportingTaskNode taskNode, final
LifecycleState lifecycleState) {
+ final boolean cronDriven = taskNode.getSchedulingStrategy() ==
SchedulingStrategy.CRON_DRIVEN;
+ final CronExpression cronExpression;
+ final long schedulingNanos;
+ if (cronDriven) {
+ cronExpression =
parseCronExpression(taskNode.getSchedulingPeriod(), taskNode);
+ schedulingNanos = 0L;
+ } else {
+ cronExpression = null;
+ schedulingNanos =
taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS);
+ }
+
+ final String componentId = taskNode.getIdentifier();
+ final SchedulingGeneration generation;
+ synchronized (lifecycleState) {
+ generation = registerSchedulingGeneration(componentId);
+ lifecycleState.setScheduled(true);
+ }
+
+ try {
+ final Runnable reportingTaskWrapper = new
ReportingTaskWrapper(taskNode, lifecycleState,
flowController.getExtensionManager());
+ final String threadName = "Reporting Task: " + taskNode.getName();
+
+ submitTask(threadName, generation,
+ () -> runReportingTaskLoop(taskNode, reportingTaskWrapper,
schedulingNanos, cronExpression, lifecycleState, generation));
+
+ logger.info("{} started on virtual thread",
taskNode.getReportingTask());
+ } catch (final Throwable t) {
+ synchronized (lifecycleState) {
+ if (stopSchedulingGeneration(componentId, generation, true)) {
+ lifecycleState.setScheduled(false);
+ }
+ }
+
+ throw t;
+ }
+ }
+
+ @Override
+ public void unschedule(final ReportingTaskNode taskNode, final
LifecycleState lifecycleState) {
+ synchronized (lifecycleState) {
+ final SchedulingGeneration generation =
schedulingGenerations.remove(taskNode.getIdentifier());
+ if (generation != null) {
+ generation.stop(false);
+ }
+
+ lifecycleState.setScheduled(false);
+ }
+
+ logger.info("Stopped scheduling {} to run",
taskNode.getReportingTask());
+ }
+
+ private SchedulingGeneration registerSchedulingGeneration(final String
componentId) {
+ if (shutdown.get()) {
+ throw new IllegalStateException("VirtualThreadSchedulingAgent has
been shut down and cannot accept new work");
+ }
+
+ final SchedulingGeneration generation = new SchedulingGeneration();
+ final SchedulingGeneration existingGeneration =
schedulingGenerations.putIfAbsent(componentId, generation);
+ if (existingGeneration != null) {
+ throw new IllegalStateException("Component " + componentId + " is
already scheduled");
+ }
+
+ if (shutdown.get()) {
+ stopSchedulingGeneration(componentId, generation, true);
+ throw new IllegalStateException("VirtualThreadSchedulingAgent has
been shut down and cannot accept new work");
+ }
+
+ return generation;
+ }
+
+ private boolean stopSchedulingGeneration(final String componentId, final
SchedulingGeneration generation, final boolean interrupt) {
+ final boolean removed = schedulingGenerations.remove(componentId,
generation);
+ generation.stop(interrupt);
+ return removed;
+ }
+
+ private boolean isActive(final LifecycleState lifecycleState, final
SchedulingGeneration generation) {
+ return !shutdown.get() && lifecycleState.isScheduled() &&
!generation.isStopped();
+ }
+
+ private static CronExpression parseCronExpression(final String
cronSchedule, final Object component) {
+ try {
+ return CronExpression.parse(cronSchedule);
+ } catch (final RuntimeException e) {
+ throw new IllegalStateException("Cannot schedule " + component + "
to run because its scheduling period is not a valid CRON expression: " +
cronSchedule, e);
+ }
+ }
+
+ @Override
+ public void onEvent(final Connectable connectable) {
+ }
+
+ @Override
+ public synchronized void setMaxThreadCount(final int maxThreads) {
+ globalSemaphore.setMaxPermits(maxThreads);
+ logger.info("Global semaphore permits updated to {}", maxThreads);
+ }
+
+ @Override
+ public synchronized void incrementMaxThreadCount(final int toAdd) {
+ if (toAdd == 0) {
+ return;
+ }
+
+ final int currentMax = globalSemaphore.getMaxPermits();
+ final int newMax = currentMax + toAdd;
+ if (newMax < 1) {
+ throw new IllegalStateException("Cannot remove " + (-toAdd) + "
permits from global semaphore because there are only " + currentMax + " permits
available");
+ }
+
+ globalSemaphore.setMaxPermits(newMax);
+ }
+
+ @Override
+ public void setAdministrativeYieldDuration(final String duration) {
+ this.adminYieldNanos = FormatUtils.getTimeDuration(duration,
TimeUnit.NANOSECONDS);
+ this.adminYieldDuration = duration;
+ }
+
+ @Override
+ public String getAdministrativeYieldDuration() {
+ return adminYieldDuration;
+ }
+
+ @Override
+ public long getAdministrativeYieldDuration(final TimeUnit timeUnit) {
+ return timeUnit.convert(adminYieldNanos, TimeUnit.NANOSECONDS);
+ }
+
+ DynamicSemaphore getGlobalSemaphore() {
+ return globalSemaphore;
+ }
+
+ int getRunningThreadCount() {
+ return runningThreadCount.get();
+ }
+
+ boolean isShutdown() {
+ return shutdown.get();
+ }
+
+ /**
+ * @return number of component invocations currently holding global permits
+ */
+ public int getActiveThreadCount() {
+ return globalSemaphore.getInUsePermits();
+ }
+
+ private void runSchedulingLoop(final Connectable connectable, final
ConnectableTask connectableTask, final long schedulingNanos,
+ final LifecycleState lifecycleState, final
SchedulingGeneration generation, final CronExpression cronExpression) {
+ final boolean cronDriven = cronExpression != null;
+
+ OffsetDateTime nextCronSchedule = null;
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(OffsetDateTime.now(),
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no future firings;
scheduling loop will exit without invoking the component", connectable);
+ return;
+ }
+
+ final long initialDelayMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ if (initialDelayMillis > 0L) {
+
waitForDelay(TimeUnit.MILLISECONDS.toNanos(initialDelayMillis), generation);
+ }
+ }
+
+ while (true) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ return;
+ }
+
+ final InvocationResult invocationResult;
+ try {
+ invocationResult = connectableTask.invoke();
+ } finally {
+ // Interrupt status from one invocation must not carry
into the scheduling loop.
+ Thread.interrupted();
+ globalSemaphore.release();
+ }
+
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(nextCronSchedule,
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no further
firings after the current invocation; scheduling loop is exiting", connectable);
+ return;
+ }
+
+ final long sleepMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ waitForDelay(TimeUnit.MILLISECONDS.toNanos(sleepMillis),
generation);
+ } else {
+ waitForNextInvocation(connectable, schedulingNanos,
generation, invocationResult);
+ }
+ } catch (final Throwable t) {
+ if (!isActive(lifecycleState, generation)) {
+ return;
+ }
+
+ try {
+ connectable.yield(adminYieldNanos, TimeUnit.NANOSECONDS);
+ } catch (final Throwable yieldError) {
+ t.addSuppressed(yieldError);
+ }
+
+ logger.error("Unexpected error in scheduling loop for {}. Will
yield for {} and continue.", connectable, adminYieldDuration, t);
+ waitForDelay(adminYieldNanos, generation);
+ }
+ }
+ }
+
+ private void runOnce(final Connectable connectable, final ConnectableTask
connectableTask, final Callable<Future<Void>> stopCallback,
+ final LifecycleState lifecycleState, final
SchedulingGeneration generation) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ if (isActive(lifecycleState, generation)) {
+ logger.warn("Run once request for {} was not executed
because permit acquisition was interrupted", connectable);
+ } else {
+ logger.warn("Run once request for {} was not executed
because scheduling is no longer active", connectable);
+ }
+
+ return;
+ }
+
+ try {
+ connectableTask.invoke();
+ } finally {
+ globalSemaphore.release();
+ }
+ } catch (final Throwable t) {
+ logger.error("Unexpected error running {} once", connectable, t);
+ } finally {
+ try {
+ stopCallback.call();
+ } catch (final Throwable t) {
+ logger.error("Error while stopping {} after running once",
connectable, t);
+ }
+ }
+ }
+
+ private void runReportingTaskLoop(final ReportingTaskNode taskNode, final
Runnable reportingTaskWrapper, final long schedulingNanos,
+ final CronExpression cronExpression,
final LifecycleState lifecycleState, final SchedulingGeneration generation) {
+ final boolean cronDriven = cronExpression != null;
+
+ OffsetDateTime nextCronSchedule = null;
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(OffsetDateTime.now(),
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no future firings;
scheduling loop will exit without invoking the reporting task",
+ taskNode.getReportingTask());
+ return;
+ }
+
+ final long initialDelayMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ if (initialDelayMillis > 0L) {
+
waitForDelay(TimeUnit.MILLISECONDS.toNanos(initialDelayMillis), generation);
+ }
+ }
+
+ while (true) {
+ try {
+ if (!acquirePermitWithPolling(lifecycleState, generation)) {
+ return;
+ }
+
+ try {
+ reportingTaskWrapper.run();
+ } finally {
+ // Interrupt status from one invocation must not carry
into the scheduling loop.
+ Thread.interrupted();
+ globalSemaphore.release();
+ }
+
+ if (cronDriven) {
+ nextCronSchedule = getNextCronSchedule(nextCronSchedule,
cronExpression);
+ if (nextCronSchedule == null) {
+ logger.warn("CRON expression for {} has no further
firings after the current invocation; scheduling loop is exiting",
+ taskNode.getReportingTask());
+ return;
+ }
+
+ final long sleepMillis =
Math.max(nextCronSchedule.toInstant().toEpochMilli() -
System.currentTimeMillis(), 0L);
+ waitForDelay(TimeUnit.MILLISECONDS.toNanos(sleepMillis),
generation);
+ } else {
+ waitForDelay(schedulingNanos, generation);
+ }
+ } catch (final Throwable t) {
+ if (!isActive(lifecycleState, generation)) {
+ return;
+ }
+
+ logger.error("Unexpected error in scheduling loop for {}. Will
wait for {} and continue.", taskNode.getReportingTask(), adminYieldDuration, t);
+ waitForDelay(adminYieldNanos, generation);
+ }
+ }
+ }
+
+ private void waitForNextInvocation(final Connectable connectable, final
long schedulingNanos, final SchedulingGeneration generation,
+ final InvocationResult
invocationResult) {
+ final long sleepNanos;
+ final long yieldExpiration = connectable.getYieldExpiration();
+ final long yieldDelayNanos;
+ if (yieldExpiration == 0L) {
+ yieldDelayNanos = 0L;
+ } else {
+ yieldDelayNanos =
TimeUnit.MILLISECONDS.toNanos(Math.max(yieldExpiration -
System.currentTimeMillis(), 0L));
+ }
+
+ if (yieldDelayNanos > 0L) {
+ sleepNanos = Math.max(schedulingNanos, yieldDelayNanos);
+ } else if (invocationResult.isYield()) {
+ sleepNanos = noWorkYieldNanos > 0L ? noWorkYieldNanos :
schedulingNanos;
+ } else {
+ sleepNanos = schedulingNanos;
+ }
+
+ waitForDelay(sleepNanos, generation);
+ }
+
+ private boolean acquirePermitWithPolling(final LifecycleState
lifecycleState, final SchedulingGeneration generation) {
+ while (isActive(lifecycleState, generation)) {
+ try {
+ if (globalSemaphore.tryAcquire(PERMIT_POLL_INTERVAL_NANOS,
TimeUnit.NANOSECONDS)) {
+ if (isActive(lifecycleState, generation)) {
+ return true;
+ }
+
+ globalSemaphore.release();
+ return false;
+ }
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ return false;
+ }
+
+ private void waitForDelay(final long delayNanos, final
SchedulingGeneration generation) {
+ if (delayNanos <= Triggerable.MINIMUM_SCHEDULING_NANOS) {
+ return;
+ }
+
+ try {
+ generation.awaitStop(delayNanos, TimeUnit.NANOSECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static String buildThreadName(final Connectable connectable, final
int taskIndex) {
+ return connectable.getName() + "[type=" +
connectable.getComponentType() + ", id=" + connectable.getIdentifier()
Review Comment:
[GPT-5.6 Sol] Updated in 35dc8b9643f to use a formatted String.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
