diff --git a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java index 48681c77c705..12a3f01ed691 100644 --- a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java +++ b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java @@ -83,6 +83,7 @@ public class NiFiProperties extends ApplicationProperties { public static final String REMOTE_CONTENTS_CACHE_EXPIRATION = "nifi.remote.contents.cache.expiration"; public static final String ADMINISTRATIVE_YIELD_DURATION = "nifi.administrative.yield.duration"; public static final String BORED_YIELD_DURATION = "nifi.bored.yield.duration"; + public static final String SCHEDULING_STRATEGY = "nifi.scheduling.strategy"; public static final String PROCESSOR_SCHEDULING_TIMEOUT = "nifi.processor.scheduling.timeout"; public static final String BACKPRESSURE_COUNT = "nifi.queue.backpressure.count"; public static final String BACKPRESSURE_SIZE = "nifi.queue.backpressure.size"; @@ -376,6 +377,10 @@ public class NiFiProperties extends ApplicationProperties { public static final String DEFAULT_ADMINISTRATIVE_YIELD_DURATION = "30 sec"; public static final String DEFAULT_COMPONENT_STATUS_SNAPSHOT_FREQUENCY = "5 mins"; public static final String DEFAULT_BORED_YIELD_DURATION = "10 millis"; + public static final String AUTO_SCHEDULING_STRATEGY = "AUTO"; + public static final String STANDARD_SCHEDULING_STRATEGY = "STANDARD"; + public static final String VIRTUAL_SCHEDULING_STRATEGY = "VIRTUAL"; + public static final String DEFAULT_SCHEDULING_STRATEGY = AUTO_SCHEDULING_STRATEGY; public static final String DEFAULT_ZOOKEEPER_CONNECT_TIMEOUT = "3 secs"; public static final String DEFAULT_ZOOKEEPER_SESSION_TIMEOUT = "3 secs"; public static final String DEFAULT_ZOOKEEPER_ROOT_NODE = "/nifi"; @@ -1468,6 +1473,10 @@ public String getBoredYieldDuration() { return getProperty(BORED_YIELD_DURATION, DEFAULT_BORED_YIELD_DURATION); } + public String getSchedulingStrategy() { + return getProperty(SCHEDULING_STRATEGY, DEFAULT_SCHEDULING_STRATEGY); + } + public File getStateManagementConfigFile() { return new File(getProperty(STATE_MANAGEMENT_CONFIG_FILE, DEFAULT_STATE_MANAGEMENT_CONFIG_FILE)); } diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc b/nifi-docs/src/main/asciidoc/administration-guide.adoc index b415d750ef04..fa310d158b6f 100644 --- a/nifi-docs/src/main/asciidoc/administration-guide.adoc +++ b/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 according to the Maximum Timer Driven Thread Count configured in Controller Settings. |`nifi.queue.backpressure.count`|When drawing a new connection between two components, this is the default value for that connection's back pressure object threshold. The default is `10000` and the value must be an integer. |`nifi.queue.backpressure.size`|When drawing a new connection between two components, this is the default value for that connection's back pressure data size threshold. The default is `1 GB` and the value must be a data size including the unit of measure. |`nifi.authorizer.configuration.file`*|This is the location of the file that specifies how authorizers are defined. The default value is `./conf/authorizers.xml`. diff --git a/nifi-framework-api/src/main/java/org/apache/nifi/diagnostics/ThreadDumpTask.java b/nifi-framework-api/src/main/java/org/apache/nifi/diagnostics/ThreadDumpTask.java index 74228a2d0aa6..882b540dd2b8 100644 --- a/nifi-framework-api/src/main/java/org/apache/nifi/diagnostics/ThreadDumpTask.java +++ b/nifi-framework-api/src/main/java/org/apache/nifi/diagnostics/ThreadDumpTask.java @@ -16,116 +16,141 @@ */ 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); + private static final String TEMPORARY_DIRECTORY_PREFIX = ThreadDumpTask.class.getSimpleName(); + private static final String TEMPORARY_FILE_NAME = ThreadDumpTask.class.getSimpleName() + ".txt"; + @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 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()); + + return new StandardDiagnosticsDumpElement("Thread Dump", Collections.singletonList(dumpBuilder.toString())); + } + + private String captureThreadDump(final ThreadMXBean threadMXBean) { + try { + return captureHotSpotThreadDump(); + } catch (final IOException | RuntimeException | LinkageError e) { + logger.warn("Failed to capture virtual threads using the HotSpot diagnostic interface; capturing platform threads instead", e); + return capturePlatformThreadDump(threadMXBean); + } + } + + private String captureHotSpotThreadDump() throws IOException { + final HotSpotDiagnosticMXBean diagnosticMXBean = ManagementFactory.getPlatformMXBean(HotSpotDiagnosticMXBean.class); + if (diagnosticMXBean == null) { + throw new UnsupportedOperationException("HotSpot diagnostic interface is not available"); + } + + final Path tempDirectory = Files.createTempDirectory(TEMPORARY_DIRECTORY_PREFIX); + final Path tempFile = tempDirectory.resolve(TEMPORARY_FILE_NAME); + try { + diagnosticMXBean.dumpThreads(tempFile.toString(), HotSpotDiagnosticMXBean.ThreadDumpFormat.TEXT_PLAIN); + return Files.readString(tempFile); + } finally { + try { + Files.deleteIfExists(tempFile); + Files.deleteIfExists(tempDirectory); + } catch (final IOException e) { + logger.debug("Failed to delete temporary thread-dump files in {}", tempDirectory, e); } - }); + } + } + + String capturePlatformThreadDump(final ThreadMXBean threadMXBean) { + final ThreadInfo[] threadInfos = threadMXBean.dumpAllThreads(true, true); + final List sortedThreadInfos = new ArrayList<>(threadInfos.length); + Collections.addAll(sortedThreadInfos, threadInfos); + sortedThreadInfos.sort(Comparator.comparing(ThreadInfo::getThreadName, String.CASE_INSENSITIVE_ORDER)); - final StringBuilder sb = new StringBuilder(); - for (final ThreadInfo info : sortedInfos) { - sb.append("\n"); - sb.append("\"").append(info.getThreadName()).append("\" Id="); - sb.append(info.getThreadId()).append(" "); - sb.append(info.getThreadState().toString()).append(" "); + final StringBuilder dumpBuilder = new StringBuilder(); + for (final ThreadInfo threadInfo : sortedThreadInfos) { + dumpBuilder.append(System.lineSeparator()) + .append('"').append(threadInfo.getThreadName()).append("\" Id=") + .append(threadInfo.getThreadId()).append(' ') + .append(threadInfo.getThreadState()); - switch (info.getThreadState()) { + switch (threadInfo.getThreadState()) { case BLOCKED: case TIMED_WAITING: case WAITING: - sb.append(" on "); - sb.append(info.getLockInfo()); + dumpBuilder.append(" on ").append(threadInfo.getLockInfo()); + if (threadInfo.getLockOwnerName() != null) { + dumpBuilder.append(" owned by \"").append(threadInfo.getLockOwnerName()).append("\" Id=").append(threadInfo.getLockOwnerId()); + } break; default: break; } - if (info.isSuspended()) { - sb.append(" (suspended)"); - } - if (info.isInNative()) { - sb.append(" (in native code)"); + if (threadInfo.isSuspended()) { + dumpBuilder.append(" (suspended)"); } - - if (deadlockedThreadIds != null) { - for (final long id : deadlockedThreadIds) { - if (id == info.getThreadId()) { - sb.append(" ** DEADLOCKED THREAD **"); - } - } + if (threadInfo.isInNative()) { + dumpBuilder.append(" (in native code)"); } - if (monitorDeadlockThreadIds != null) { - for (final long id : monitorDeadlockThreadIds) { - if (id == info.getThreadId()) { - sb.append(" ** MONITOR-DEADLOCKED THREAD **"); + final MonitorInfo[] lockedMonitors = threadInfo.getLockedMonitors(); + for (final StackTraceElement stackTraceElement : threadInfo.getStackTrace()) { + dumpBuilder.append(System.lineSeparator()).append("\tat ").append(stackTraceElement); + for (final MonitorInfo monitorInfo : lockedMonitors) { + if (Objects.equals(monitorInfo.getLockedStackFrame(), stackTraceElement)) { + dumpBuilder.append(System.lineSeparator()).append("\t- locked ").append(monitorInfo); } } } - final StackTraceElement[] stackTraces = info.getStackTrace(); - for (final StackTraceElement element : stackTraces) { - sb.append("\n\tat ").append(element); - - final MonitorInfo[] monitors = info.getLockedMonitors(); - for (final MonitorInfo monitor : monitors) { - if (Objects.equals(monitor.getLockedStackFrame(), element)) { - sb.append("\n\t- waiting on ").append(monitor); - } + final LockInfo[] lockedSynchronizers = threadInfo.getLockedSynchronizers(); + if (lockedSynchronizers.length > 0) { + dumpBuilder.append(System.lineSeparator()).append("\tNumber of Locked Synchronizers: ").append(lockedSynchronizers.length); + for (final LockInfo lockInfo : lockedSynchronizers) { + dumpBuilder.append(System.lineSeparator()).append("\t- locked ").append(lockInfo); } } + dumpBuilder.append(System.lineSeparator()); + } - final LockInfo[] lockInfos = info.getLockedSynchronizers(); - if (lockInfos.length > 0) { - sb.append("\n\t"); - sb.append("Number of Locked Synchronizers: ").append(lockInfos.length); - for (final LockInfo lockInfo : lockInfos) { - sb.append("\n\t- ").append(lockInfo.toString()); - } - } + return dumpBuilder.toString(); + } - sb.append("\n"); + private void appendDeadlockedThreadIds(final StringBuilder dumpBuilder, final String heading, final long[] threadIds) { + if (threadIds == null || threadIds.length == 0) { + return; } - if (deadlockedThreadIds != null && deadlockedThreadIds.length > 0) { - sb.append("\n\nDEADLOCK DETECTED!"); - sb.append("\nThe following thread IDs are deadlocked:"); - for (final long id : deadlockedThreadIds) { - sb.append("\n").append(id); - } - } + dumpBuilder.append(System.lineSeparator()).append(System.lineSeparator()).append(heading) + .append(System.lineSeparator()).append("The following thread IDs are deadlocked:"); - if (monitorDeadlockThreadIds != null && monitorDeadlockThreadIds.length > 0) { - sb.append("\n\nMONITOR DEADLOCK DETECTED!"); - sb.append("\nThe following thread IDs are deadlocked:"); - for (final long id : monitorDeadlockThreadIds) { - sb.append("\n").append(id); - } + for (final long threadId : threadIds) { + dumpBuilder.append(System.lineSeparator()).append(threadId); } - - return new StandardDiagnosticsDumpElement("Thread Dump", Collections.singletonList(sb.toString())); } } diff --git a/nifi-framework-api/src/test/java/org/apache/nifi/diagnostics/ThreadDumpTaskTest.java b/nifi-framework-api/src/test/java/org/apache/nifi/diagnostics/ThreadDumpTaskTest.java new file mode 100644 index 000000000000..7b2207259165 --- /dev/null +++ b/nifi-framework-api/src/test/java/org/apache/nifi/diagnostics/ThreadDumpTaskTest.java @@ -0,0 +1,105 @@ +/* + * 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.diagnostics; + +import org.junit.jupiter.api.Test; + +import java.lang.management.ManagementFactory; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class ThreadDumpTaskTest { + + @Test + void testCaptureDumpIncludesPlatformAndVirtualThreads() throws InterruptedException { + final String platformThreadName = "thread-dump-platform-thread"; + final String virtualThreadName = "thread-dump-virtual-thread"; + final CountDownLatch threadsStarted = new CountDownLatch(2); + final CountDownLatch releaseThreads = new CountDownLatch(1); + final Runnable task = () -> { + threadsStarted.countDown(); + try { + releaseThreads.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }; + + final Thread platformThread = Thread.ofPlatform().name(platformThreadName).start(task); + final Thread virtualThread = Thread.ofVirtual().name(virtualThreadName).start(task); + + try { + assertTrue(threadsStarted.await(2, TimeUnit.SECONDS)); + + final DiagnosticsDumpElement dumpElement = new ThreadDumpTask().captureDump(false); + final String threadDump = dumpElement.getDetails().getFirst(); + + assertTrue(threadDump.contains(platformThreadName)); + assertTrue(threadDump.contains(virtualThreadName)); + } finally { + releaseThreads.countDown(); + platformThread.join(2_000L); + virtualThread.join(2_000L); + } + } + + @Test + void testPlatformFallbackIncludesBlockedMonitor() throws InterruptedException { + final Object monitor = new Object(); + final CountDownLatch monitorAcquired = new CountDownLatch(1); + final CountDownLatch releaseMonitor = new CountDownLatch(1); + final Thread lockOwner = Thread.ofPlatform().name("thread-dump-lock-owner").start(() -> { + synchronized (monitor) { + monitorAcquired.countDown(); + try { + releaseMonitor.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }); + + try { + assertTrue(monitorAcquired.await(2, TimeUnit.SECONDS)); + final Thread blockedThread = Thread.ofPlatform().name("thread-dump-blocked-thread").start(() -> { + synchronized (monitor) { + } + }); + + try { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2L); + while (blockedThread.getState() != Thread.State.BLOCKED && System.nanoTime() < deadline) { + Thread.onSpinWait(); + } + assertEquals(Thread.State.BLOCKED, blockedThread.getState()); + + final String threadDump = new ThreadDumpTask().capturePlatformThreadDump(ManagementFactory.getThreadMXBean()); + final String blockedThreadHeader = "\"thread-dump-blocked-thread\" Id=" + blockedThread.threadId() + " BLOCKED on "; + assertTrue(threadDump.contains(blockedThreadHeader)); + } finally { + releaseMonitor.countDown(); + blockedThread.join(2_000L); + } + } finally { + releaseMonitor.countDown(); + lockOwner.join(2_000L); + } + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java index 2c042fd816ed..ca158e0b10e3 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java @@ -574,7 +574,7 @@ public void yield() { @Override public void yield(final long period, final TimeUnit timeUnit) { final long yieldMillis = TimeUnit.MILLISECONDS.convert(period, timeUnit); - yieldExpiration.set(Math.max(yieldExpiration.get(), System.currentTimeMillis() + yieldMillis)); + yieldExpiration.accumulateAndGet(System.currentTimeMillis() + yieldMillis, Math::max); processScheduler.yield(this); } @@ -585,7 +585,16 @@ public void yield(final long period, final TimeUnit timeUnit) { */ @Override public long getYieldExpiration() { - return yieldExpiration.get(); + final long expiration = yieldExpiration.get(); + if (expiration == 0L) { + return 0L; + } + + if (expiration > System.currentTimeMillis()) { + return expiration; + } + + return yieldExpiration.compareAndSet(expiration, 0L) ? 0L : yieldExpiration.get(); } @Override @@ -1575,7 +1584,9 @@ public List getActiveThreads(final ThreadDetails threadDetails final long activeMillis = now - timestamp; final ThreadInfo threadInfo = threadInfoMap.get(thread.threadId()); - final String stackTrace = ThreadUtils.createStackTrace(threadInfo, threadDetails.getDeadlockedThreadIds(), threadDetails.getMonitorDeadlockThreadIds()); + final String stackTrace = threadInfo == null + ? ThreadUtils.createStackTrace(thread) + : ThreadUtils.createStackTrace(threadInfo, threadDetails.getDeadlockedThreadIds(), threadDetails.getMonitorDeadlockThreadIds()); final ActiveThreadInfo activeThreadInfo = new ActiveThreadInfo(thread.getName(), stackTrace, activeMillis, activeTask.isTerminated()); threadList.add(activeThreadInfo); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/ThreadUtils.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/ThreadUtils.java index b6d6e3a8679f..2f3a14385a65 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/ThreadUtils.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/util/ThreadUtils.java @@ -24,6 +24,19 @@ public class ThreadUtils { + public static String createStackTrace(final Thread thread) { + final StringBuilder stringBuilder = new StringBuilder(); + stringBuilder.append('"').append(thread.getName()).append("\" Id=") + .append(thread.threadId()).append(' ') + .append(thread.getState()); + + for (final StackTraceElement stackTraceElement : thread.getStackTrace()) { + stringBuilder.append(System.lineSeparator()).append("\tat ").append(stackTraceElement); + } + + return stringBuilder.append(System.lineSeparator()).toString(); + } + public static String createStackTrace(final ThreadInfo threadInfo, final long[] deadlockedThreadIds, final long[] monitorDeadlockThreadIds) { final StringBuilder sb = new StringBuilder(); sb.append("\"").append(threadInfo.getThreadName()).append("\" Id="); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/StandardProcessorNodeTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/StandardProcessorNodeTest.java new file mode 100644 index 000000000000..eec1c95fbebf --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/StandardProcessorNodeTest.java @@ -0,0 +1,104 @@ +/* + * 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.bundle.BundleCoordinate; +import org.apache.nifi.components.validation.ValidationTrigger; +import org.apache.nifi.components.validation.VerifiableComponentFactory; +import org.apache.nifi.controller.service.ControllerServiceProvider; +import org.apache.nifi.nar.ExtensionManager; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.ProcessSessionFactory; +import org.apache.nifi.processor.Processor; +import org.apache.nifi.util.NoOpProcessor; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class StandardProcessorNodeTest { + + @Test + void testYieldExpiration() { + final ProcessScheduler processScheduler = mock(ProcessScheduler.class); + final Processor processor = new NoOpProcessor(); + final StandardProcessorNode processorNode = createProcessorNode(processor, processScheduler); + + processorNode.yield(0L, TimeUnit.MILLISECONDS); + assertEquals(0L, processorNode.getYieldExpiration()); + + processorNode.yield(1L, TimeUnit.DAYS); + final long expiration = processorNode.getYieldExpiration(); + assertTrue(expiration > System.currentTimeMillis()); + + processorNode.yield(1L, TimeUnit.SECONDS); + assertEquals(expiration, processorNode.getYieldExpiration()); + } + + @Test + void testGetActiveThreadsIncludesVirtualThread() throws InterruptedException { + final String threadName = "virtual-processor-task"; + final CountDownLatch invocationStarted = new CountDownLatch(1); + final CountDownLatch releaseInvocation = new CountDownLatch(1); + final Processor processor = new NoOpProcessor() { + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) { + invocationStarted.countDown(); + try { + releaseInvocation.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }; + final StandardProcessorNode processorNode = createProcessorNode(processor, mock(ProcessScheduler.class)); + final ProcessSessionFactory sessionFactory = mock(ProcessSessionFactory.class); + when(sessionFactory.createSession()).thenReturn(mock(ProcessSession.class)); + final Thread virtualThread = Thread.ofVirtual().name(threadName).start(() -> processorNode.onTrigger(mock(ProcessContext.class), sessionFactory)); + + try { + assertTrue(invocationStarted.await(2, TimeUnit.SECONDS)); + + final List activeThreads = processorNode.getActiveThreads(ThreadDetails.capture()); + + assertEquals(1, activeThreads.size()); + assertEquals(threadName, activeThreads.getFirst().getThreadName()); + assertTrue(activeThreads.getFirst().getStackTrace().contains(threadName)); + assertTrue(activeThreads.getFirst().getStackTrace().contains(StandardProcessorNodeTest.class.getName())); + } finally { + releaseInvocation.countDown(); + virtualThread.join(2_000L); + } + + assertFalse(virtualThread.isAlive()); + } + + private StandardProcessorNode createProcessorNode(final Processor processor, final ProcessScheduler processScheduler) { + final LoggableComponent loggableProcessor = new LoggableComponent<>(processor, BundleCoordinate.UNKNOWN_COORDINATE, null); + return new StandardProcessorNode(loggableProcessor, "processor", mock(ValidationContextFactory.class), processScheduler, + mock(ControllerServiceProvider.class), mock(ReloadComponent.class), mock(VerifiableComponentFactory.class), + mock(ExtensionManager.class), mock(ValidationTrigger.class)); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java index 8398907cfa1f..dc8b76528a60 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/scheduling/LifecycleStateTest.java @@ -26,6 +26,8 @@ import java.lang.ref.Reference; import java.lang.ref.WeakReference; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -34,6 +36,23 @@ class LifecycleStateTest { private static final long GARBAGE_COLLECTION_TIMEOUT_MILLIS = 5_000L; + @Test + void testLastStopTimeAdvancesOnEveryStop() { + final LifecycleState lifecycleState = new LifecycleState("component-id"); + lifecycleState.setScheduled(true); + final long initialStopTime = lifecycleState.getLastStopTime(); + + lifecycleState.setScheduled(false); + final long firstStopTime = lifecycleState.getLastStopTime(); + assertTrue(firstStopTime > initialStopTime); + + lifecycleState.setScheduled(true); + assertEquals(firstStopTime, lifecycleState.getLastStopTime()); + + lifecycleState.setScheduled(false); + assertTrue(lifecycleState.getLastStopTime() > firstStopTime); + } + /** * Verifies that a Session created by an ActiveProcessSessionFactory is rolled back when the * LifecycleState is terminated, even when the only strong reference to the factory has been released. diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java index 3a30b189c716..2e527d0abb10 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/scheduling/LifecycleState.java @@ -110,7 +110,13 @@ public synchronized void setScheduled(final boolean scheduled) { mustCallOnStoppedMethods.set(true); if (!scheduled) { - lastStopTime = System.currentTimeMillis(); + final long previousStopTime = lastStopTime; + long nextStopTime = System.currentTimeMillis(); + if (nextStopTime <= previousStopTime) { + nextStopTime = previousStopTime + 1L; + } + + lastStopTime = nextStopTime; } } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java index 32cfc3b7ae53..69ac6f697482 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java @@ -137,6 +137,7 @@ import org.apache.nifi.controller.scheduling.StandardLifecycleStateManager; import org.apache.nifi.controller.scheduling.StandardProcessScheduler; import org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent; +import org.apache.nifi.controller.scheduling.VirtualThreadSchedulingAgent; import org.apache.nifi.controller.serialization.FlowSerializationException; import org.apache.nifi.controller.serialization.FlowSerializer; import org.apache.nifi.controller.serialization.FlowSynchronizationException; @@ -292,9 +293,12 @@ public class FlowController implements ReportingTaskProvider, FlowAnalysisRulePr public static final long DEFAULT_GRACEFUL_SHUTDOWN_SECONDS = 10; private static final String ZOOKEEPER_STATE_PROVIDER_SERVER_CLASS = "org.apache.nifi.controller.state.providers.zookeeper.server.ZooKeeperStateProviderServer"; + private static final int MINIMUM_JAVA_VERSION_FOR_AUTOMATIC_VIRTUAL_THREAD_SCHEDULING = 25; + private static final int FRAMEWORK_TASK_THREAD_COUNT = 8; private final AtomicInteger maxTimerDrivenThreads; private final AtomicReference timerDrivenEngineRef; + private final VirtualThreadSchedulingAgent virtualThreadSchedulingAgent; private final ContentRepository contentRepository; private final FlowFileRepository flowFileRepository; @@ -551,7 +555,10 @@ private FlowController( stateManagerProvider.enableClusterProvider(); } - timerDrivenEngineRef = new AtomicReference<>(new FlowEngine(maxTimerDrivenThreads.get(), "Timer-Driven Process")); + final boolean virtualThreadSchedulingEnabled = isVirtualThreadSchedulingEnabled(nifiProperties.getSchedulingStrategy(), Runtime.version().feature()); + final int flowEngineThreadCount = virtualThreadSchedulingEnabled ? FRAMEWORK_TASK_THREAD_COUNT : maxTimerDrivenThreads.get(); + final String flowEngineName = virtualThreadSchedulingEnabled ? "Framework Task" : "Timer-Driven Process"; + timerDrivenEngineRef = new AtomicReference<>(new FlowEngine(flowEngineThreadCount, flowEngineName)); final FlowFileRepository flowFileRepo = createFlowFileRepository(nifiProperties, extensionManager, resourceClaimManager); flowFileRepository = flowFileRepo; @@ -671,10 +678,20 @@ private FlowController( flowAnalyzer.initialize(controllerServiceProvider); } - final CronSchedulingAgent cronSchedulingAgent = new CronSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory); - final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory, this.nifiProperties); - processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, timerDrivenAgent); - processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, cronSchedulingAgent); + if (virtualThreadSchedulingEnabled) { + this.virtualThreadSchedulingAgent = new VirtualThreadSchedulingAgent(this, repositoryContextFactory, this.nifiProperties, maxTimerDrivenThreads.get()); + processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, virtualThreadSchedulingAgent); + processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, virtualThreadSchedulingAgent); + LOG.info("Component scheduling configured to use virtual threads with a maximum of {} concurrent tasks", maxTimerDrivenThreads.get()); + } else { + this.virtualThreadSchedulingAgent = null; + + final TimerDrivenSchedulingAgent timerDrivenAgent = new TimerDrivenSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory, this.nifiProperties); + final CronSchedulingAgent cronSchedulingAgent = new CronSchedulingAgent(this, timerDrivenEngineRef.get(), repositoryContextFactory); + processScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, timerDrivenAgent); + processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, cronSchedulingAgent); + LOG.info("Component scheduling configured to use a platform thread pool of {} threads", maxTimerDrivenThreads.get()); + } startConnectablesAfterInitialization = new HashSet<>(); startRemoteGroupPortsAfterInitialization = new HashSet<>(); @@ -891,6 +908,23 @@ private FlowController( : Optional.empty(); } + private static boolean isVirtualThreadSchedulingEnabled(final String schedulingStrategy, final int javaFeatureVersion) { + if (NiFiProperties.AUTO_SCHEDULING_STRATEGY.equals(schedulingStrategy)) { + return javaFeatureVersion >= MINIMUM_JAVA_VERSION_FOR_AUTOMATIC_VIRTUAL_THREAD_SCHEDULING; + } + + if (NiFiProperties.VIRTUAL_SCHEDULING_STRATEGY.equals(schedulingStrategy)) { + return true; + } + + if (NiFiProperties.STANDARD_SCHEDULING_STRATEGY.equals(schedulingStrategy)) { + return false; + } + + throw new IllegalArgumentException("Unsupported value [%s] configured for property [%s]. Supported values are AUTO, STANDARD, and VIRTUAL." + .formatted(schedulingStrategy, NiFiProperties.SCHEDULING_STRATEGY)); + } + @Override public Authorizable getParentAuthorizable() { return null; @@ -1875,7 +1909,8 @@ public Authorizer getAuthorizer() { public boolean isTerminated() { this.readLock.lock(); try { - return null == this.timerDrivenEngineRef.get() || this.timerDrivenEngineRef.get().isTerminated(); + final boolean timerDrivenEngineTerminated = timerDrivenEngineRef.get() == null || timerDrivenEngineRef.get().isTerminated(); + return timerDrivenEngineTerminated && (virtualThreadSchedulingAgent == null || virtualThreadSchedulingAgent.isTerminated()); } finally { this.readLock.unlock("isTerminated"); } @@ -1950,18 +1985,33 @@ public void shutdown(final boolean kill) { if (kill) { this.timerDrivenEngineRef.get().shutdownNow(); + + if (virtualThreadSchedulingAgent != null) { + virtualThreadSchedulingAgent.shutdown(); + } + LOG.info("Initiated immediate shutdown of flow controller..."); } else { this.timerDrivenEngineRef.get().shutdown(); + + if (virtualThreadSchedulingAgent != null) { + virtualThreadSchedulingAgent.shutdownGracefully(); + } + LOG.info("Initiated graceful shutdown of flow controller...waiting up to {} seconds", gracefulShutdownSeconds); } try { - // Give thread pool up to the configured amount of time to finish, but no less than 2 seconds, - // in order to allow for a more graceful shutdown. - final long millisToWait = Math.max(2000, shutdownEnd - System.currentTimeMillis()); - this.timerDrivenEngineRef.get().awaitTermination(millisToWait, TimeUnit.MILLISECONDS); - } catch (final InterruptedException ie) { + final long terminationEnd = Math.max(shutdownEnd, System.currentTimeMillis() + 2_000L); + final long timerDrivenMillisToWait = Math.max(0L, terminationEnd - System.currentTimeMillis()); + this.timerDrivenEngineRef.get().awaitTermination(timerDrivenMillisToWait, TimeUnit.MILLISECONDS); + + if (virtualThreadSchedulingAgent != null) { + final long virtualThreadMillisToWait = Math.max(0L, terminationEnd - System.currentTimeMillis()); + virtualThreadSchedulingAgent.awaitTermination(virtualThreadMillisToWait, TimeUnit.MILLISECONDS); + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); LOG.info("Interrupted while waiting for controller termination."); } @@ -1971,7 +2021,8 @@ public void shutdown(final boolean kill) { LOG.warn("Unable to shut down FlowFileRepository", t); } - if (this.timerDrivenEngineRef.get().isTerminated()) { + final boolean virtualThreadSchedulingTerminated = virtualThreadSchedulingAgent == null || virtualThreadSchedulingAgent.isTerminated(); + if (this.timerDrivenEngineRef.get().isTerminated() && virtualThreadSchedulingTerminated) { LOG.info("Controller has been terminated successfully."); } else { LOG.warn("Controller hasn't terminated properly. There exists an uninterruptible thread that " @@ -2194,39 +2245,38 @@ public int getMaxTimerDrivenThreadCount() { } public int getActiveTimerDrivenThreadCount() { - return timerDrivenEngineRef.get().getActiveCount(); - } - - public void setMaxTimerDrivenThreadCount(final int maxThreadCount) { - writeLock.lock(); - try { - setMaxThreadCount(maxThreadCount, "Timer Driven", this.timerDrivenEngineRef.get(), this.maxTimerDrivenThreads); - } finally { - writeLock.unlock("setMaxTimerDrivenThreadCount"); + if (virtualThreadSchedulingAgent == null) { + return timerDrivenEngineRef.get().getActiveCount(); } + + return virtualThreadSchedulingAgent.getActiveThreadCount(); } - /** - * Updates the number of threads that can be simultaneously used for executing processors. - * This method must be called while holding the write lock! - * - * @param maxThreadCount Requested new thread pool size - * @param poolName Thread Pool Name - * @param engine Flow Engine executor or null when terminated - * @param maxThreads Internal tracker for Maximum Threads - */ - private void setMaxThreadCount(final int maxThreadCount, final String poolName, final FlowEngine engine, final AtomicInteger maxThreads) { + public void setMaxTimerDrivenThreadCount(final int maxThreadCount) { if (maxThreadCount < 1) { throw new IllegalArgumentException("Cannot set max number of threads to less than 1"); } - maxThreads.getAndSet(maxThreadCount); - if (engine == null) { - LOG.debug("[{}] Engine not found: Maximum Thread Count not updated", poolName); - } else { - final int previousCorePoolSize = engine.getCorePoolSize(); - engine.setCorePoolSize(maxThreadCount); - LOG.info("[{}] Maximum Thread Count updated [{}] previous [{}]", poolName, maxThreadCount, previousCorePoolSize); + writeLock.lock(); + try { + final int previousMax = maxTimerDrivenThreads.getAndSet(maxThreadCount); + + if (virtualThreadSchedulingAgent != null) { + virtualThreadSchedulingAgent.setMaxThreadCount(maxThreadCount); + } else { + final FlowEngine engine = timerDrivenEngineRef.get(); + if (engine == null) { + LOG.debug("Timer-Driven Engine not found: Maximum Thread Count not updated"); + } else { + final int previousCorePoolSize = engine.getCorePoolSize(); + engine.setCorePoolSize(maxThreadCount); + LOG.debug("Timer-Driven Engine core pool size updated [{}] previous [{}]", maxThreadCount, previousCorePoolSize); + } + } + + LOG.info("Maximum Timer-Driven Thread Count updated [{}] previous [{}]", maxThreadCount, previousMax); + } finally { + writeLock.unlock("setMaxTimerDrivenThreadCount"); } } @@ -2868,7 +2918,7 @@ public GroupStatusCounts getGroupStatusCounts(final ProcessGroup group) { } public int getActiveThreadCount() { - return timerDrivenEngineRef.get().getActiveCount(); + return getActiveTimerDrivenThreadCount(); } // diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/DynamicSemaphore.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/DynamicSemaphore.java new file mode 100644 index 000000000000..39584a48199a --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/DynamicSemaphore.java @@ -0,0 +1,100 @@ +/* + * 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 java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +/** + * Semaphore with a configurable maximum permit count and fair waiter ordering. + */ +public class DynamicSemaphore { + + private final ResizableSemaphore semaphore; + private volatile int maxPermits; + + public DynamicSemaphore(final int permits) { + if (permits < 1) { + throw new IllegalArgumentException("Permits must be at least 1"); + } + + this.maxPermits = permits; + this.semaphore = new ResizableSemaphore(permits); + } + + public void acquire() throws InterruptedException { + semaphore.acquire(); + } + + public boolean tryAcquire(final long timeout, final TimeUnit timeUnit) throws InterruptedException { + return semaphore.tryAcquire(timeout, timeUnit); + } + + public void release() { + semaphore.release(); + } + + /** + * Adjusts the maximum permit count without interrupting current permit holders. + * + * @param newMaxPermits maximum permits, at least one + */ + public synchronized void setMaxPermits(final int newMaxPermits) { + if (newMaxPermits < 1) { + throw new IllegalArgumentException("Max permits must be at least 1"); + } + + final int delta = newMaxPermits - this.maxPermits; + this.maxPermits = newMaxPermits; + + if (delta > 0) { + semaphore.release(delta); + } else if (delta < 0) { + semaphore.reducePermits(-delta); + } + } + + public int getMaxPermits() { + return maxPermits; + } + + public int availablePermits() { + return semaphore.availablePermits(); + } + + /** + * Returns the number of acquired permits. The result can exceed the configured maximum + * while a reduced permit limit waits for current holders to release permits. + * + * @return acquired permit count + */ + public synchronized int getInUsePermits() { + return maxPermits - semaphore.availablePermits(); + } + + private static class ResizableSemaphore extends Semaphore { + + ResizableSemaphore(final int permits) { + super(permits, true); + } + + @Override + protected void reducePermits(final int reduction) { + super.reducePermits(reduction); + } + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java index 6f8744d728e0..8d9e6dd331a3 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/StandardProcessScheduler.java @@ -769,8 +769,7 @@ private synchronized void startConnectable(final Connectable connectable) { lifecycleState.clearTerminationFlag(); - // Schedule the component to be triggered, unless the engine is stateless. For stateless engine, we let the stateless - // framework take care of triggering components. + // Stateless components are driven by the stateless framework, so no scheduling agent is involved. if (connectable.getProcessGroup().resolveExecutionEngine() != ExecutionEngine.STATELESS) { getSchedulingAgent(connectable).schedule(connectable, lifecycleState); } diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java new file mode 100644 index 000000000000..911b700a0c35 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgent.java @@ -0,0 +1,631 @@ +/* + * 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 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, generation::isRunning); + 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> 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, generation::isRunning); + 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(), generation::isRunning); + 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> 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 "%s[type=%s, id=%s, group=%s] task %d".formatted(connectable.getName(), connectable.getComponentType(), connectable.getIdentifier(), + connectable.getProcessGroup().getName(), taskIndex); + } + + private void submitTask(final String threadName, final SchedulingGeneration generation, final Runnable task) { + final Runnable trackedTask = () -> { + final Thread currentThread = Thread.currentThread(); + currentThread.setName(threadName); + generation.addThread(currentThread); + runningThreadCount.incrementAndGet(); + + try { + if (!shutdown.get() && !generation.isStopped()) { + task.run(); + } + } finally { + runningThreadCount.decrementAndGet(); + generation.removeThread(currentThread); + } + }; + + executorService.execute(trackedTask); + } + + private static OffsetDateTime getNextCronSchedule(final OffsetDateTime currentSchedule, final CronExpression cronExpression) { + final OffsetDateTime now = OffsetDateTime.now(); + return cronExpression.next(now.isAfter(currentSchedule) ? now : currentSchedule); + } + + private static class SchedulingGeneration { + private final CountDownLatch stopSignal = new CountDownLatch(1); + private final Set threads = ConcurrentHashMap.newKeySet(); + private final AtomicBoolean interruptRequested = new AtomicBoolean(); + + void addThread(final Thread thread) { + threads.add(thread); + + if (interruptRequested.get()) { + thread.interrupt(); + } + } + + void removeThread(final Thread thread) { + threads.remove(thread); + } + + void stop(final boolean interrupt) { + if (interrupt) { + interruptRequested.set(true); + } + + stopSignal.countDown(); + + if (interruptRequested.get()) { + for (final Thread thread : threads) { + thread.interrupt(); + } + } + } + + boolean isStopped() { + return stopSignal.getCount() == 0L; + } + + boolean isRunning() { + return !isStopped(); + } + + void awaitStop(final long timeout, final TimeUnit timeUnit) throws InterruptedException { + stopSignal.await(timeout, timeUnit); + } + } +} 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 a7c7dfc1799b..7c94e8a61630 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 @@ -25,7 +25,6 @@ import org.apache.nifi.controller.ScheduledState; import org.apache.nifi.controller.lifecycle.TaskTerminationAwareStateManager; import org.apache.nifi.controller.metrics.ProcessSessionEvent; -import org.apache.nifi.controller.queue.FlowFileQueue; import org.apache.nifi.controller.repository.ActiveProcessSessionFactory; import org.apache.nifi.controller.repository.BatchingSessionFactory; import org.apache.nifi.controller.repository.RepositoryContext; @@ -57,6 +56,7 @@ import java.io.IOException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BooleanSupplier; /** * Continually runs a {@link Connectable} component as long as the component has work to do. @@ -65,6 +65,10 @@ public class ConnectableTask { private static final Logger logger = LoggerFactory.getLogger(ConnectableTask.class); + private static final BooleanSupplier SCHEDULING_GENERATION_ALWAYS_ACTIVE = () -> true; + private static final InvocationResult NOT_PRIMARY_NODE_YIELD_RESULT = InvocationResult.yield("This node is not the primary node"); + private static final InvocationResult NO_WORK_YIELD_RESULT = InvocationResult.yield("No work to do"); + private static final InvocationResult BACKPRESSURE_YIELD_RESULT = InvocationResult.yield("Backpressure Applied"); private final SchedulingAgent schedulingAgent; private final Connectable connectable; @@ -74,13 +78,20 @@ public class ConnectableTask { private final FlowController flowController; private final int numRelationships; private final StatsTracker statsTracker; + private final BooleanSupplier schedulingGenerationActive; - public ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable, - final FlowController flowController, final RepositoryContextFactory contextFactory, final LifecycleState lifecycleState) { + public ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable, final FlowController flowController, + final RepositoryContextFactory contextFactory, final LifecycleState lifecycleState) { + this(schedulingAgent, connectable, flowController, contextFactory, lifecycleState, SCHEDULING_GENERATION_ALWAYS_ACTIVE); + } + + public ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable, final FlowController flowController, + final RepositoryContextFactory contextFactory, final LifecycleState lifecycleState, final BooleanSupplier schedulingGenerationActive) { this.schedulingAgent = schedulingAgent; this.connectable = connectable; this.lifecycleState = lifecycleState; + this.schedulingGenerationActive = schedulingGenerationActive; this.numRelationships = connectable.getRelationships().size(); this.flowController = flowController; @@ -115,10 +126,9 @@ private boolean isRunOnCluster(final FlowController flowController) { } private boolean isYielded() { - // after one yield period, the scheduling agent could call this again when - // yieldExpiration == currentTime, and we don't want that to still be considered 'yielded' - // so this uses ">" instead of ">=" - return connectable.getYieldExpiration() > System.currentTimeMillis(); + // Equality means that the yield has expired. + final long yieldExpiration = connectable.getYieldExpiration(); + return yieldExpiration > 0L && yieldExpiration > System.currentTimeMillis(); } /** @@ -135,33 +145,31 @@ private boolean isYielded() { * @return true if there is work to do, otherwise false */ private boolean isWorkToDo() { - boolean hasNonLoopConnection = Connectables.hasNonLoopConnection(connectable); - if (connectable.getConnectableType() == ConnectableType.FUNNEL) { // Handle Funnel as a special case because it will never be a 'source' component, // and also its outgoing connections can not be terminated. // Incoming FlowFiles from other components, and at least one outgoing connection are required. return connectable.hasIncomingConnection() - && hasNonLoopConnection && !connectable.getConnections().isEmpty() + && Connectables.hasNonLoopConnection(connectable) && Connectables.flowFilesQueued(connectable); } - final boolean isSourceComponent = connectable.isTriggerWhenEmpty() - // No input connections - || !connectable.hasIncomingConnection() - // Every incoming connection loops back to itself, no inputs from other components - || !hasNonLoopConnection; + if (connectable.isTriggerWhenEmpty() || !connectable.hasIncomingConnection()) { + return true; + } - // If it is not a 'source' component, it requires a FlowFile to process. - return isSourceComponent || Connectables.flowFilesQueued(connectable); + return !Connectables.hasNonLoopConnection(connectable) || Connectables.flowFilesQueued(connectable); } private boolean isBackPressureEngaged() { - return connectable.getIncomingConnections().stream() - .filter(con -> con.getSource() == connectable) - .map(Connection::getFlowFileQueue) - .anyMatch(FlowFileQueue::isFull); + for (final Connection connection : connectable.getIncomingConnections()) { + if (connection.getSource() == connectable && connection.getFlowFileQueue().isFull()) { + return true; + } + } + + return false; } public InvocationResult invoke() { @@ -186,20 +194,20 @@ public InvocationResult invoke() { } else { logger.debug("Will not trigger {} because this is not the primary node", connectable); } - return InvocationResult.yield("This node is not the primary node"); + return NOT_PRIMARY_NODE_YIELD_RESULT; } // Make sure processor has work to do. if (!isWorkToDo()) { logger.debug("Yielding {} because it has no work to do", connectable); - return InvocationResult.yield("No work to do"); + return NO_WORK_YIELD_RESULT; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { logger.debug("Yielding {} because Backpressure is Applied", connectable); - return InvocationResult.yield("Backpressure Applied"); + return BACKPRESSURE_YIELD_RESULT; } } @@ -221,7 +229,14 @@ public InvocationResult invoke() { } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); - lifecycleState.incrementActiveThreadCount(activeSessionFactory); + final boolean activeThreadCountIncremented; + synchronized (lifecycleState) { + activeThreadCountIncremented = schedulingGenerationActive.getAsBoolean() && lifecycleState.tryIncrementActiveThreadCount(activeSessionFactory); + } + if (!activeThreadCountIncremented) { + logger.debug("Will not trigger {} because it is no longer scheduled", connectable); + return InvocationResult.DO_NOT_YIELD; + } final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ReportingTaskWrapper.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ReportingTaskWrapper.java index aea99dce7e35..4c4fd8fe0da4 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ReportingTaskWrapper.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/tasks/ReportingTaskWrapper.java @@ -26,23 +26,39 @@ import org.apache.nifi.processor.StandardComponentLog; import org.apache.nifi.util.ReflectionUtils; +import java.util.function.BooleanSupplier; + public class ReportingTaskWrapper implements Runnable { + private static final BooleanSupplier SCHEDULING_GENERATION_ALWAYS_ACTIVE = () -> true; + private final ReportingTaskNode taskNode; private final LifecycleState lifecycleState; private final ExtensionManager extensionManager; + private final BooleanSupplier schedulingGenerationActive; public ReportingTaskWrapper(final ReportingTaskNode taskNode, final LifecycleState lifecycleState, final ExtensionManager extensionManager) { + this(taskNode, lifecycleState, extensionManager, SCHEDULING_GENERATION_ALWAYS_ACTIVE); + } + + public ReportingTaskWrapper(final ReportingTaskNode taskNode, final LifecycleState lifecycleState, final ExtensionManager extensionManager, + final BooleanSupplier schedulingGenerationActive) { this.taskNode = taskNode; this.lifecycleState = lifecycleState; this.extensionManager = extensionManager; + this.schedulingGenerationActive = schedulingGenerationActive; } @Override - public synchronized void run() { - if (!lifecycleState.tryIncrementActiveThreadCount(null)) { + public void run() { + final boolean activeThreadCountIncremented; + synchronized (lifecycleState) { + activeThreadCountIncremented = schedulingGenerationActive.getAsBoolean() && lifecycleState.tryIncrementActiveThreadCount(null); + } + if (!activeThreadCountIncremented) { return; } + try (final NarCloseable ignored = NarCloseable.withComponentNarLoader(extensionManager, taskNode.getReportingTask().getClass(), taskNode.getIdentifier())) { taskNode.getReportingTask().onTrigger(taskNode.getReportingContext()); } catch (final Throwable t) { diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java index 6cfb081dc59c..fc10ef301c9b 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/diagnostics/bootstrap/tasks/NiFiPropertiesDiagnosticTask.java @@ -48,6 +48,7 @@ public class NiFiPropertiesDiagnosticTask implements DiagnosticTask { "nifi.flowfile.repository.always.sync", "nifi.components.status.snapshot.frequency", "nifi.bored.yield.duration", + "nifi.scheduling.strategy", "nifi.queue.swap.threshold", "nifi.security.identity.mapping.pattern.dn", "nifi.security.identity.mapping.value.dn", diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/DynamicSemaphoreTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/DynamicSemaphoreTest.java new file mode 100644 index 000000000000..ba7834bf8606 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/DynamicSemaphoreTest.java @@ -0,0 +1,184 @@ +/* + * 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.junit.jupiter.api.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class DynamicSemaphoreTest { + + @Test + void testConstructorRejectsZeroPermits() { + assertThrows(IllegalArgumentException.class, () -> new DynamicSemaphore(0)); + } + + @Test + void testInitialPermitCount() { + final DynamicSemaphore semaphore = new DynamicSemaphore(5); + assertEquals(5, semaphore.getMaxPermits()); + assertEquals(5, semaphore.availablePermits()); + } + + @Test + void testAcquireAndRelease() throws InterruptedException { + final DynamicSemaphore semaphore = new DynamicSemaphore(3); + semaphore.acquire(); + assertEquals(2, semaphore.availablePermits()); + semaphore.acquire(); + assertEquals(1, semaphore.availablePermits()); + semaphore.release(); + assertEquals(2, semaphore.availablePermits()); + semaphore.release(); + assertEquals(3, semaphore.availablePermits()); + } + + @Test + void testConcurrencyBoundedByPermits() throws InterruptedException { + final int permits = 2; + final int threadCount = 5; + final DynamicSemaphore semaphore = new DynamicSemaphore(permits); + final AtomicInteger concurrentCount = new AtomicInteger(0); + final AtomicInteger maxObservedConcurrency = new AtomicInteger(0); + final CountDownLatch allStarted = new CountDownLatch(threadCount); + final CountDownLatch allDone = new CountDownLatch(threadCount); + + for (int i = 0; i < threadCount; i++) { + Thread.ofVirtual().start(() -> { + try { + allStarted.countDown(); + semaphore.acquire(); + try { + final int current = concurrentCount.incrementAndGet(); + maxObservedConcurrency.accumulateAndGet(current, Math::max); + Thread.sleep(50); + } finally { + concurrentCount.decrementAndGet(); + semaphore.release(); + } + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + allDone.countDown(); + } + }); + } + + assertTrue(allDone.await(5, TimeUnit.SECONDS)); + assertTrue(maxObservedConcurrency.get() <= permits, + "Max concurrency " + maxObservedConcurrency.get() + " exceeded permit count " + permits); + } + + @Test + void testSetMaxPermitsIncrease() throws InterruptedException { + final DynamicSemaphore semaphore = new DynamicSemaphore(2); + semaphore.acquire(); + semaphore.acquire(); + assertEquals(0, semaphore.availablePermits()); + + semaphore.setMaxPermits(5); + assertEquals(5, semaphore.getMaxPermits()); + assertEquals(3, semaphore.availablePermits()); + } + + @Test + void testSetMaxPermitsDecrease() { + final DynamicSemaphore semaphore = new DynamicSemaphore(5); + assertEquals(5, semaphore.availablePermits()); + + semaphore.setMaxPermits(2); + assertEquals(2, semaphore.getMaxPermits()); + assertEquals(2, semaphore.availablePermits()); + } + + @Test + void testSetMaxPermitsDecreaseWhileHeld() throws InterruptedException { + final DynamicSemaphore semaphore = new DynamicSemaphore(5); + semaphore.acquire(); + semaphore.acquire(); + semaphore.acquire(); + assertEquals(2, semaphore.availablePermits()); + + semaphore.setMaxPermits(2); + assertEquals(2, semaphore.getMaxPermits()); + assertTrue(semaphore.availablePermits() <= 0, + "Available permits should be non-positive when more permits are held than the new max"); + + semaphore.release(); + semaphore.release(); + semaphore.release(); + assertEquals(2, semaphore.availablePermits()); + } + + @Test + void testSetMaxPermitsRejectsZero() { + final DynamicSemaphore semaphore = new DynamicSemaphore(5); + assertThrows(IllegalArgumentException.class, () -> semaphore.setMaxPermits(0)); + } + + @Test + void testResizeUnblocksWaitingThreads() throws InterruptedException { + final DynamicSemaphore semaphore = new DynamicSemaphore(1); + semaphore.acquire(); + + final CountDownLatch threadBlocked = new CountDownLatch(1); + final CountDownLatch threadAcquired = new CountDownLatch(1); + + Thread.ofVirtual().start(() -> { + try { + threadBlocked.countDown(); + semaphore.acquire(); + threadAcquired.countDown(); + semaphore.release(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + assertTrue(threadBlocked.await(1, TimeUnit.SECONDS)); + semaphore.setMaxPermits(2); + assertTrue(threadAcquired.await(2, TimeUnit.SECONDS)); + + semaphore.release(); + } + + @Test + void testGetInUsePermitsReflectsAcquireAndRelease() throws InterruptedException { + final DynamicSemaphore semaphore = new DynamicSemaphore(4); + assertEquals(0, semaphore.getInUsePermits()); + + semaphore.acquire(); + assertEquals(1, semaphore.getInUsePermits()); + + semaphore.acquire(); + semaphore.acquire(); + assertEquals(3, semaphore.getInUsePermits()); + + semaphore.release(); + assertEquals(2, semaphore.getInUsePermits()); + + semaphore.release(); + semaphore.release(); + assertEquals(0, semaphore.getInUsePermits()); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java index e45c13857e51..437fb43cc851 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/TestStandardProcessScheduler.java @@ -28,6 +28,7 @@ import org.apache.nifi.components.validation.ValidationTrigger; import org.apache.nifi.components.validation.VerifiableComponentFactory; import org.apache.nifi.connectable.Connectable; +import org.apache.nifi.connectable.Funnel; import org.apache.nifi.controller.AbstractControllerService; import org.apache.nifi.controller.ConfigurationContext; import org.apache.nifi.controller.ExtensionBuilder; @@ -125,7 +126,9 @@ import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -257,6 +260,55 @@ public void testReportingTaskDoesntKeepRunningAfterStop() throws InterruptedExce "After unscheduling Reporting Task, task ran an additional " + attemptsAfterStop + " times"); } + @Test + @Timeout(30) + public void testReportingTaskRunsWithVirtualThreadSchedulingAgent() throws InterruptedException, InitializationException { + verifyReportingTaskRunsWithSchedulingAgent(true); + } + + @Test + @Timeout(30) + public void testReportingTaskRunsWithTimerDrivenSchedulingAgent() throws InterruptedException, InitializationException { + verifyReportingTaskRunsWithSchedulingAgent(false); + } + + private void verifyReportingTaskRunsWithSchedulingAgent(final boolean virtualThreads) throws InterruptedException, InitializationException { + final FlowController flowController = mock(FlowController.class); + when(flowController.getExtensionManager()).thenReturn(extensionManager); + when(flowController.getReloadComponent()).thenReturn(mock(ReloadComponent.class)); + + final FlowEngine flowEngine = new FlowEngine(2, "Scheduling Agent Unit Test", true); + final StandardProcessScheduler realScheduler = new StandardProcessScheduler(flowEngine, extensionManager, + flowController, () -> serviceProvider, mock(ReloadComponent.class), stateMgrProvider, nifiProperties, new StandardLifecycleStateManager()); + + final RepositoryContextFactory contextFactory = mock(RepositoryContextFactory.class); + final SchedulingAgent schedulingAgent = virtualThreads + ? new VirtualThreadSchedulingAgent(flowController, contextFactory, nifiProperties, 10) + : new TimerDrivenSchedulingAgent(flowController, flowEngine, contextFactory, nifiProperties); + realScheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, schedulingAgent); + + final TestReportingTask task = new TestReportingTask(); + task.failOnScheduled.set(false); + final ReportingInitializationContext config = new StandardReportingInitializationContext(UUID.randomUUID().toString(), "SchedulingAgentTest", SchedulingStrategy.TIMER_DRIVEN, + "10 millis", mock(ComponentLog.class), null, KerberosConfig.NOT_CONFIGURED, null); + + task.initialize(config); + final LoggableComponent loggableTask = new LoggableComponent<>(task, systemBundle.getBundleDetails().getCoordinate(), mock(TerminationAwareLogger.class)); + final ReportingTaskNode taskNode = new StandardReportingTaskNode(loggableTask, UUID.randomUUID().toString(), flowController, realScheduler, + new StandardValidationContextFactory(null), mock(ReloadComponent.class), extensionManager, new SynchronousValidationTrigger()); + taskNode.setSchedulingPeriod("10 millis"); + taskNode.performValidation(); + + realScheduler.schedule(taskNode); + + try { + assertTrue(task.triggered.await(5, TimeUnit.SECONDS)); + } finally { + realScheduler.unschedule(taskNode); + realScheduler.shutdown(); + } + } + @Test @Timeout(60) public void testDisableControllerServiceWithProcessorTryingToStartUsingIt() throws InterruptedException, ExecutionException { @@ -304,19 +356,21 @@ public class TestReportingTask extends AbstractReportingTask { private final AtomicBoolean failOnScheduled = new AtomicBoolean(true); private final AtomicInteger onScheduleAttempts = new AtomicInteger(0); private final AtomicInteger triggerCount = new AtomicInteger(0); + private final CountDownLatch triggered = new CountDownLatch(1); @OnScheduled public void onScheduled() { onScheduleAttempts.incrementAndGet(); if (failOnScheduled.get()) { - throw new RuntimeException("Intentional Exception for testing purposes"); + throw new IllegalStateException("Intentional Exception for testing purposes"); } } @Override public void onTrigger(final ReportingContext context) { triggerCount.getAndIncrement(); + triggered.countDown(); } } @@ -631,6 +685,26 @@ public void testProcessorStopWaitsForSchedulingAgentUnschedule() throws Exceptio assertEquals(ScheduledState.STOPPED, processorNode.getPhysicalScheduledState()); } + @Test + public void testFunnelCanStartAfterSchedulingFailure() { + final SchedulingAgent schedulingAgent = mock(SchedulingAgent.class); + final IllegalStateException schedulingFailure = new IllegalStateException("Scheduling failed"); + doThrow(schedulingFailure).doNothing().when(schedulingAgent).schedule(any(Connectable.class), any(LifecycleState.class)); + scheduler.setSchedulingAgent(SchedulingStrategy.TIMER_DRIVEN, schedulingAgent); + + final Funnel funnel = mock(Funnel.class); + when(funnel.getIdentifier()).thenReturn("funnel"); + when(funnel.getProcessGroup()).thenReturn(rootGroup); + when(funnel.getScheduledState()).thenReturn(ScheduledState.STOPPED); + when(funnel.getSchedulingStrategy()).thenReturn(SchedulingStrategy.TIMER_DRIVEN); + + assertSame(schedulingFailure, assertThrows(IllegalStateException.class, () -> scheduler.startFunnel(funnel))); + scheduler.startFunnel(funnel); + scheduler.startFunnel(funnel); + + verify(schedulingAgent, times(2)).schedule(any(Connectable.class), any(LifecycleState.class)); + } + private Void scheduleAgent(final boolean scheduled, final LifecycleState lifecycleState, final CountDownLatch callbackStarted, final Semaphore callbackRelease) { lifecycleState.setScheduled(scheduled); @@ -853,14 +927,7 @@ public void testTerminateProcessorRollsBackRetainedSessionWhenNoActiveThreads() } /** - * Verifies that the stop background poll loop in {@code StandardProcessorNode.stop()} exits cleanly - * once {@link LifecycleState#terminate()} has been invoked, instead of rescheduling itself every - * 100ms forever in the component lifecycle thread pool. - * - * Without the fix, {@code LifecycleState.terminate()} resets the active thread count to zero, which - * the poll loop interprets as "still waiting for threads to drain" (it is comparing against 1, which - * represents the stop background thread itself), so it keeps rescheduling and leaks one polling task - * per terminated processor. + * Verifies that the stop background poll loop exits when the lifecycle state is terminated. */ @Test @Timeout(30) diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgentTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgentTest.java new file mode 100644 index 000000000000..47428da4b8ea --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/scheduling/VirtualThreadSchedulingAgentTest.java @@ -0,0 +1,849 @@ +/* + * 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(); + private static final String CRON_SCHEDULE = "* * * * * ?"; + + @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 inheritedValue = new InheritableThreadLocal<>(); + final ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader(); + final ClassLoader lifecycleClassLoader = new ClassLoader(originalClassLoader) { + }; + final AtomicReference observedClassLoader = new AtomicReference<>(); + final AtomicReference 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 testUnscheduleAfterPermitAcquiredPreventsInvocation() throws InterruptedException { + final AtomicInteger invocationCount = new AtomicInteger(); + final CountDownLatch invocationCheckStarted = new CountDownLatch(1); + final CountDownLatch releaseInvocationCheck = new CountDownLatch(1); + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.TIMER_DRIVEN, invocationCount, new CountDownLatch(0)); + when(connectable.getYieldExpiration()).thenAnswer(invocation -> { + invocationCheckStarted.countDown(); + releaseInvocationCheck.await(); + return 0L; + }); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + + try { + assertTrue(invocationCheckStarted.await(2, TimeUnit.SECONDS)); + unscheduleConnectable(connectable, lifecycleState); + } finally { + releaseInvocationCheck.countDown(); + } + + waitForRunningThreadCount(0, 2, TimeUnit.SECONDS); + + 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(CRON_SCHEDULE); + when(connectable.evaluateParameters(eq(CRON_SCHEDULE))).thenReturn(CRON_SCHEDULE); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + + assertTrue(atLeastOneInvocation.await(3, TimeUnit.SECONDS), + "CRON-scheduled connectable should have invoked at least once within 3 seconds"); + + unscheduleConnectable(connectable, lifecycleState); + waitForRunningThreadCount(0, 2, TimeUnit.SECONDS); + } + + @Test + void testCronDrivenReportingTaskIsScheduled() throws InterruptedException { + final CountDownLatch invocationCompleted = new CountDownLatch(1); + final ReportingTask reportingTask = mock(ReportingTask.class); + doAnswer(invocation -> { + invocationCompleted.countDown(); + return null; + }).when(reportingTask).onTrigger(any()); + + final ReportingTaskNode taskNode = mock(ReportingTaskNode.class); + when(taskNode.getSchedulingStrategy()).thenReturn(SchedulingStrategy.CRON_DRIVEN); + when(taskNode.getSchedulingPeriod()).thenReturn(CRON_SCHEDULE); + when(taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS)) + .thenThrow(new IllegalArgumentException("CRON expression cannot be parsed as a time duration")); + when(taskNode.getReportingTask()).thenReturn(reportingTask); + when(taskNode.getReportingContext()).thenReturn(mock(ReportingContext.class)); + when(taskNode.getIdentifier()).thenReturn(COMPONENT_ID); + when(taskNode.getName()).thenReturn("TestReporter"); + when(flowController.getExtensionManager()).thenReturn(extensionManager); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + lifecycleState.setScheduled(true); + agent.schedule(taskNode, lifecycleState); + + try { + assertTrue(invocationCompleted.await(3, TimeUnit.SECONDS)); + } finally { + lifecycleState.setScheduled(false); + agent.unschedule(taskNode, lifecycleState); + waitForRunningThreadCount(0, 2, TimeUnit.SECONDS); + } + } + + @Test + void testReportingTaskContinuesAfterInterruptStatusSet() throws InterruptedException { + final AtomicInteger invocationCount = new AtomicInteger(); + final CountDownLatch secondInvocation = new CountDownLatch(1); + final ReportingTask reportingTask = mock(ReportingTask.class); + doAnswer(invocation -> { + if (invocationCount.incrementAndGet() == 1) { + Thread.currentThread().interrupt(); + } else { + secondInvocation.countDown(); + } + + return null; + }).when(reportingTask).onTrigger(any()); + + final ReportingTaskNode taskNode = mock(ReportingTaskNode.class); + when(taskNode.getSchedulingStrategy()).thenReturn(SchedulingStrategy.TIMER_DRIVEN); + when(taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(100L)); + when(taskNode.getReportingTask()).thenReturn(reportingTask); + when(taskNode.getReportingContext()).thenReturn(mock(ReportingContext.class)); + when(taskNode.getIdentifier()).thenReturn(COMPONENT_ID); + when(taskNode.getName()).thenReturn("TestReporter"); + when(flowController.getExtensionManager()).thenReturn(extensionManager); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + lifecycleState.setScheduled(true); + agent.schedule(taskNode, lifecycleState); + + try { + assertTrue(secondInvocation.await(2, TimeUnit.SECONDS)); + } finally { + lifecycleState.setScheduled(false); + agent.unschedule(taskNode, lifecycleState); + } + } + + @Test + void testCronConnectableExitsCleanlyWhenNoFutureFirings() throws InterruptedException { + final String unreachableCron = "0 0 0 30 2 ?"; + final AtomicInteger invocationCount = new AtomicInteger(0); + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.CRON_DRIVEN, invocationCount, new CountDownLatch(0)); + when(connectable.getSchedulingPeriod()).thenReturn(unreachableCron); + when(connectable.evaluateParameters(eq(unreachableCron))).thenReturn(unreachableCron); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + + Thread.sleep(500L); + + assertEquals(0, invocationCount.get()); + assertEquals(MAX_THREADS, agent.getGlobalSemaphore().availablePermits()); + + unscheduleConnectable(connectable, lifecycleState); + } + + @Test + void testRapidStopStartDoesNotLeakSchedulingThreads() throws InterruptedException { + final AtomicReference firstSchedulingThread = new AtomicReference<>(); + 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.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(500L)); + doAnswer(invocation -> { + final int currentInvocation = invocationCount.incrementAndGet(); + if (currentInvocation == 1) { + firstSchedulingThread.set(Thread.currentThread()); + 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)); + + unscheduleConnectable(connectable, lifecycleState); + scheduleConnectable(connectable, lifecycleState); + assertTrue(secondInvocation.await(2, TimeUnit.SECONDS)); + + firstSchedulingThread.get().join(2_000L); + assertFalse(firstSchedulingThread.get().isAlive()); + } finally { + unscheduleConnectable(connectable, lifecycleState); + waitForRunningThreadCount(0, 2, TimeUnit.SECONDS); + } + } + + private void scheduleConnectable(final Connectable connectable, final LifecycleState lifecycleState) { + lifecycleState.setScheduled(true); + agent.schedule(connectable, lifecycleState); + } + + private void unscheduleConnectable(final Connectable connectable, final LifecycleState lifecycleState) { + lifecycleState.setScheduled(false); + agent.unschedule(connectable, lifecycleState); + } + + private void waitForRunningThreadCount(final int expectedCount, final long timeout, final TimeUnit timeUnit) throws InterruptedException { + final long deadline = System.nanoTime() + timeUnit.toNanos(timeout); + while (agent.getRunningThreadCount() != expectedCount && System.nanoTime() < deadline) { + Thread.sleep(10L); + } + assertEquals(expectedCount, agent.getRunningThreadCount()); + } + + private Connectable createMockedConnectable(final int maxConcurrentTasks, final SchedulingStrategy schedulingStrategy, + final AtomicInteger invocationCount, final CountDownLatch invocationLatch) { + final Connectable connectable = mock(Connectable.class); + when(connectable.getIdentifier()).thenReturn(COMPONENT_ID); + when(connectable.getName()).thenReturn("TestProcessor"); + when(connectable.getMaxConcurrentTasks()).thenReturn(maxConcurrentTasks); + when(connectable.getIncomingConnections()).thenReturn(Collections.emptyList()); + when(connectable.getRelationships()).thenReturn(Collections.emptySet()); + when(connectable.getSchedulingPeriod(TimeUnit.MILLISECONDS)).thenReturn(100L); + when(connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(100L)); + when(connectable.getYieldExpiration()).thenReturn(0L); + when(connectable.getSchedulingStrategy()).thenReturn(schedulingStrategy); + when(connectable.isTriggerWhenEmpty()).thenReturn(true); + when(connectable.isIsolated()).thenReturn(false); + when(connectable.getRunDuration(TimeUnit.NANOSECONDS)).thenReturn(0L); + when(connectable.isSessionBatchingSupported()).thenReturn(false); + when(connectable.getScheduledState()).thenReturn(ScheduledState.RUNNING); + + final Processor runnableComponent = mock(Processor.class); + when(connectable.getRunnableComponent()).thenReturn(runnableComponent); + + doAnswer(invocation -> { + invocationCount.incrementAndGet(); + invocationLatch.countDown(); + return null; + }).when(connectable).onTrigger(any(), any()); + + final ProcessGroup processGroup = mock(ProcessGroup.class); + when(processGroup.getName()).thenReturn("RootGroup"); + when(processGroup.getParent()).thenReturn(null); + when(connectable.getProcessGroup()).thenReturn(processGroup); + + when(flowController.getStateManagerProvider()).thenReturn(stateManagerProvider); + when(stateManagerProvider.getStateManager(eq(COMPONENT_ID))).thenReturn(stateManager); + when(flowController.getGarbageCollectionLog()).thenReturn(garbageCollectionLog); + when(flowController.getPerformanceTrackingPercentage()).thenReturn(0); + when(flowController.getExtensionManager()).thenReturn(extensionManager); + + final RepositoryContext repositoryContext = mock(RepositoryContext.class); + when(repositoryContext.isRelationshipAvailabilitySatisfied(0)).thenReturn(true); + final FlowFileEventRepository flowFileEventRepository = mock(FlowFileEventRepository.class); + when(repositoryContext.getFlowFileEventRepository()).thenReturn(flowFileEventRepository); + when(contextFactory.newProcessContext(eq(connectable), any(AtomicLong.class))).thenReturn(repositoryContext); + + return connectable; + } + + @Test + void testScheduleRollsBackScheduledFlagOnFailure() { + final Connectable connectable = createMockedConnectable(2, SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0)); + when(connectable.getMaxConcurrentTasks()).thenThrow(new IllegalStateException("Simulated failure during schedule")); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + final IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> agent.schedule(connectable, lifecycleState)); + assertEquals("Simulated failure during schedule", thrown.getMessage()); + assertFalse(lifecycleState.isScheduled()); + assertEquals(0, agent.getRunningThreadCount()); + } + + @Test + void testScheduleOnceRollsBackScheduledFlagOnFailure() { + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0)); + when(connectable.getProcessGroup()).thenThrow(new IllegalStateException("Simulated failure building thread name")); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + assertThrows(IllegalStateException.class, () -> agent.scheduleOnce(connectable, lifecycleState, () -> null)); + assertFalse(lifecycleState.isScheduled()); + } + + @Test + void testScheduleReportingTaskRollsBackScheduledFlagOnFailure() { + final ReportingTaskNode taskNode = mock(ReportingTaskNode.class); + when(taskNode.getSchedulingStrategy()).thenReturn(SchedulingStrategy.TIMER_DRIVEN); + when(taskNode.getSchedulingPeriod(TimeUnit.NANOSECONDS)).thenReturn(TimeUnit.MILLISECONDS.toNanos(50L)); + when(taskNode.getIdentifier()).thenReturn(COMPONENT_ID); + when(taskNode.getName()).thenThrow(new IllegalStateException("Simulated failure building thread name")); + when(flowController.getExtensionManager()).thenReturn(extensionManager); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + assertThrows(IllegalStateException.class, () -> agent.schedule(taskNode, lifecycleState)); + assertFalse(lifecycleState.isScheduled()); + } + + @Test + void testShutdownInterruptsRunningVirtualThreads() 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(); + try { + releaseInvocation.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + + return null; + }).when(connectable).onTrigger(any(), any()); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + + assertTrue(invocationStarted.await(5, TimeUnit.SECONDS)); + assertTrue(agent.getRunningThreadCount() >= 1); + + agent.shutdown(); + + assertTrue(agent.awaitTermination(5, TimeUnit.SECONDS)); + assertEquals(0, agent.getRunningThreadCount()); + assertTrue(agent.isShutdown()); + } + + @Test + void testShutdownPreventsInvocationAfterInterruptedProcessorReturns() throws InterruptedException { + final AtomicInteger invocationCount = new AtomicInteger(); + final CountDownLatch firstInvocationStarted = new CountDownLatch(1); + final CountDownLatch releaseFirstInvocation = new CountDownLatch(1); + final CountDownLatch secondInvocationStarted = new CountDownLatch(1); + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0)); + doAnswer(invocation -> { + final int currentInvocation = invocationCount.incrementAndGet(); + if (currentInvocation == 1) { + firstInvocationStarted.countDown(); + try { + releaseFirstInvocation.await(); + } catch (final InterruptedException ignored) { + } + } else if (currentInvocation == 2) { + secondInvocationStarted.countDown(); + } + return null; + }).when(connectable).onTrigger(any(), any()); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + assertTrue(firstInvocationStarted.await(2, TimeUnit.SECONDS)); + + agent.shutdown(); + + assertFalse(secondInvocationStarted.await(500, TimeUnit.MILLISECONDS)); + assertTrue(agent.awaitTermination(5, TimeUnit.SECONDS)); + } + + @Test + void testGracefulShutdownWaitsForRunningInvocation() throws InterruptedException { + final CountDownLatch invocationStarted = new CountDownLatch(1); + final CountDownLatch releaseInvocation = new CountDownLatch(1); + final CountDownLatch secondInvocationStarted = new CountDownLatch(1); + final AtomicInteger invocationCount = new AtomicInteger(); + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0)); + doAnswer(invocation -> { + if (invocationCount.incrementAndGet() == 1) { + invocationStarted.countDown(); + releaseInvocation.await(); + } else { + secondInvocationStarted.countDown(); + } + return null; + }).when(connectable).onTrigger(any(), any()); + + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + scheduleConnectable(connectable, lifecycleState); + assertTrue(invocationStarted.await(2, TimeUnit.SECONDS)); + + agent.shutdownGracefully(); + + assertFalse(agent.awaitTermination(100, TimeUnit.MILLISECONDS)); + releaseInvocation.countDown(); + assertTrue(agent.awaitTermination(2, TimeUnit.SECONDS)); + assertFalse(secondInvocationStarted.await(100, TimeUnit.MILLISECONDS)); + } + + @Test + void testScheduleAfterShutdownFailsFast() { + agent.shutdown(); + + final Connectable connectable = createMockedConnectable(1, SchedulingStrategy.TIMER_DRIVEN, new AtomicInteger(), new CountDownLatch(0)); + final LifecycleState lifecycleState = new LifecycleState(COMPONENT_ID); + + assertThrows(IllegalStateException.class, () -> agent.schedule(connectable, lifecycleState)); + assertFalse(lifecycleState.isScheduled()); + } + + @Test + void testShutdownIsIdempotent() { + agent.shutdown(); + agent.shutdown(); + assertTrue(agent.isShutdown()); + assertEquals(0, agent.getRunningThreadCount()); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/ReportingTaskWrapperTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/ReportingTaskWrapperTest.java new file mode 100644 index 000000000000..0e7b20efb527 --- /dev/null +++ b/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 { + + @Test + void testRunDoesNotHoldWrapperMonitorDuringReportingTaskInvocation() { + final ReportingTask reportingTask = mock(ReportingTask.class); + final ReportingTaskNode taskNode = mock(ReportingTaskNode.class); + final ReportingContext reportingContext = mock(ReportingContext.class); + final ExtensionManager extensionManager = mock(ExtensionManager.class); + final LifecycleState lifecycleState = new LifecycleState("reporting-task"); + final AtomicBoolean callbackInvoked = new AtomicBoolean(); + final AtomicBoolean wrapperMonitorHeld = new AtomicBoolean(); + final AtomicReference wrapperReference = new AtomicReference<>(); + + when(taskNode.getReportingTask()).thenReturn(reportingTask); + when(taskNode.getReportingContext()).thenReturn(reportingContext); + when(taskNode.getIdentifier()).thenReturn("reporting-task"); + doAnswer(invocation -> { + callbackInvoked.set(true); + wrapperMonitorHeld.set(Thread.holdsLock(wrapperReference.get())); + return null; + }).when(reportingTask).onTrigger(any(ReportingContext.class)); + + final ReportingTaskWrapper wrapper = new ReportingTaskWrapper(taskNode, lifecycleState, extensionManager); + wrapperReference.set(wrapper); + lifecycleState.setScheduled(true); + + wrapper.run(); + + assertTrue(callbackInvoked.get()); + assertFalse(wrapperMonitorHeld.get()); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java index 9329eb174bbc..a41da4b0c9ad 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/tasks/TestConnectableTask.java @@ -25,6 +25,7 @@ import org.apache.nifi.controller.FlowController; import org.apache.nifi.controller.GarbageCollectionLog; import org.apache.nifi.controller.ProcessorNode; +import org.apache.nifi.controller.ScheduledState; import org.apache.nifi.controller.queue.FlowFileQueue; import org.apache.nifi.controller.repository.FlowFileEventRepository; import org.apache.nifi.controller.repository.RepositoryContext; @@ -33,6 +34,7 @@ import org.apache.nifi.controller.scheduling.RepositoryContextFactory; import org.apache.nifi.controller.scheduling.SchedulingAgent; import org.apache.nifi.controller.status.FlowFileAvailability; +import org.apache.nifi.nar.ExtensionManager; import org.apache.nifi.processor.Processor; import org.junit.jupiter.api.Test; @@ -40,11 +42,14 @@ import java.util.Collections; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +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.Mockito.any; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -53,6 +58,7 @@ public class TestConnectableTask { private ConnectableTask createTask(final Connectable connectable) { final FlowController flowController = mock(FlowController.class); when(flowController.getStateManagerProvider()).thenReturn(mock(StateManagerProvider.class)); + when(flowController.getExtensionManager()).thenReturn(mock(ExtensionManager.class)); final RepositoryContext repoContext = mock(StandardRepositoryContext.class); when(repoContext.getFlowFileEventRepository()).thenReturn(mock(FlowFileEventRepository.class)); @@ -62,10 +68,38 @@ private ConnectableTask createTask(final Connectable connectable) { final RepositoryContextFactory contextFactory = mock(RepositoryContextFactory.class); when(contextFactory.newProcessContext(any(Connectable.class), any(AtomicLong.class))).thenReturn(repoContext); - final LifecycleState scheduleState = new LifecycleState(connectable.getIdentifier()); + final LifecycleState lifecycleState = new LifecycleState(connectable.getIdentifier()); + lifecycleState.setScheduled(true); return new ConnectableTask(mock(SchedulingAgent.class), connectable, - flowController, contextFactory, scheduleState); + flowController, contextFactory, lifecycleState); + } + + @Test + public void testInvokeDoesNotHoldTaskMonitorDuringProcessorInvocation() { + final ProcessorNode processorNode = mock(ProcessorNode.class); + final Processor processor = mock(Processor.class); + final AtomicBoolean processorInvoked = new AtomicBoolean(); + final AtomicBoolean taskMonitorHeld = new AtomicBoolean(); + final AtomicReference taskReference = new AtomicReference<>(); + when(processorNode.getIdentifier()).thenReturn("processor-id"); + when(processorNode.getRunnableComponent()).thenReturn(processor); + when(processorNode.getRelationships()).thenReturn(Collections.emptySet()); + when(processorNode.getIncomingConnections()).thenReturn(Collections.emptyList()); + when(processorNode.getScheduledState()).thenReturn(ScheduledState.RUNNING); + doAnswer(invocation -> { + processorInvoked.set(true); + taskMonitorHeld.set(Thread.holdsLock(taskReference.get())); + return null; + }).when(processorNode).onTrigger(any(), any()); + + final ConnectableTask task = createTask(processorNode); + taskReference.set(task); + + task.invoke(); + + assertTrue(processorInvoked.get()); + assertFalse(taskMonitorHeld.get()); } @Test diff --git a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml index 79877e885a08..0e4c65bd0ece 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml +++ b/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml @@ -35,6 +35,7 @@ 500 ms 30 sec 10 millis + AUTO 10000 1 GB diff --git a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties index a7c8256a02bc..bbe41a671358 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties +++ b/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties @@ -27,6 +27,7 @@ nifi.flowservice.writedelay.interval=${nifi.flowservice.writedelay.interval} nifi.administrative.yield.duration=${nifi.administrative.yield.duration} # If a component has no work to do (is "bored"), how long should we wait before checking again for work? nifi.bored.yield.duration=${nifi.bored.yield.duration} +nifi.scheduling.strategy=${nifi.scheduling.strategy} nifi.queue.backpressure.count=${nifi.queue.backpressure.count} nifi.queue.backpressure.size=${nifi.queue.backpressure.size} diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/scheduling/VirtualThreadStartStopCycleIT.java b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/scheduling/VirtualThreadStartStopCycleIT.java new file mode 100644 index 000000000000..37afeb7a3a66 --- /dev/null +++ b/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. + */ +class VirtualThreadStartStopCycleIT extends NiFiSystemIT { + + private static final int START_STOP_CYCLES = 25; + + @Override + protected Map getNifiPropertiesOverrides() { + return Map.of("nifi.scheduling.strategy", "VIRTUAL"); + } + + @Test + @Timeout(value = 5, unit = TimeUnit.MINUTES) + void testRepeatedStartStopProducesExpectedQueueCount() throws NiFiClientException, IOException, InterruptedException { + final ProcessorEntity generate = getClientUtil().createProcessor("GenerateFlowFile"); + final ProcessorEntity terminate = getClientUtil().createProcessor("TerminateFlowFile"); + final ConnectionEntity generateToTerminate = getClientUtil().createConnection(generate, terminate, "success"); + + getClientUtil().updateConnectionBackpressure(generateToTerminate, 10_000L, 10_000_000L); + + final Map generateProperties = new HashMap<>(); + generateProperties.put("File Size", "0 B"); + generateProperties.put("Batch Size", "1"); + generateProperties.put("Max FlowFiles", "1"); + generateProperties.put("State Scope", "LOCAL"); + getClientUtil().updateProcessorProperties(generate, generateProperties); + + final ProcessorConfigDTO generateConfig = new ProcessorConfigDTO(); + generateConfig.setSchedulingPeriod("0 sec"); + generateConfig.setConcurrentlySchedulableTaskCount(1); + final ProcessorEntity configuredGenerate = getClientUtil().updateProcessorConfig(generate, generateConfig); + + final ProcessorConfigDTO terminateConfig = new ProcessorConfigDTO(); + terminateConfig.setSchedulingPeriod("0 sec"); + terminateConfig.setConcurrentlySchedulableTaskCount(1); + final ProcessorEntity configuredTerminate = getClientUtil().updateProcessorConfig(terminate, terminateConfig); + + for (int cycle = 1; cycle <= START_STOP_CYCLES; cycle++) { + getClientUtil().startProcessor(configuredGenerate); + waitForQueueCount(generateToTerminate, cycle); + getClientUtil().stopProcessor(configuredGenerate); + + assertEquals(cycle, getConnectionQueueSize(generateToTerminate.getId())); + } + + assertEquals(START_STOP_CYCLES, getConnectionQueueSize(generateToTerminate.getId())); + + getClientUtil().startProcessor(configuredTerminate); + + try { + waitForQueueCount(generateToTerminate, 0); + } finally { + getClientUtil().stopProcessor(configuredTerminate); + } + + assertEquals(0, getConnectionQueueSize(generateToTerminate.getId())); + } +} diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties index 9282db83eb42..80754ed38999 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties +++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties @@ -26,6 +26,7 @@ nifi.flowservice.writedelay.interval=500 ms nifi.administrative.yield.duration=100 millis # If a component has no work to do (is "bored"), how long should we wait before checking again for work? nifi.bored.yield.duration=10 millis +nifi.scheduling.strategy=VIRTUAL nifi.queue.backpressure.count=10000 nifi.queue.backpressure.size=1 GB diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties index d843563c423b..5e511ebea8b2 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties +++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties @@ -26,6 +26,7 @@ nifi.flowservice.writedelay.interval=500 ms nifi.administrative.yield.duration=100 millis # If a component has no work to do (is "bored"), how long should we wait before checking again for work? nifi.bored.yield.duration=10 millis +nifi.scheduling.strategy=VIRTUAL nifi.queue.backpressure.count=10000 nifi.queue.backpressure.size=1 GB diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties index d55f0a2d4faa..24f05a2ab9d9 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties +++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties @@ -26,6 +26,7 @@ nifi.flowservice.writedelay.interval=500 ms nifi.administrative.yield.duration=100 millis # If a component has no work to do (is "bored"), how long should we wait before checking again for work? nifi.bored.yield.duration=10 millis +nifi.scheduling.strategy=AUTO nifi.queue.backpressure.count=10000 nifi.queue.backpressure.size=1 GB diff --git a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties index bdd285d4bfd6..2f7c6dfa8de4 100644 --- a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties +++ b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties @@ -30,6 +30,7 @@ nifi.flowservice.writedelay.interval=500 ms nifi.administrative.yield.duration=100 millis # If a component has no work to do (is "bored"), how long should we wait before checking again for work? nifi.bored.yield.duration=10 millis +nifi.scheduling.strategy=VIRTUAL nifi.queue.backpressure.count=10000 nifi.queue.backpressure.size=1 GB