NIFI-15862: Moved Processors to using Virtual Threads with a Semaphor…

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));
     }

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`.

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<ThreadInfo> sortedInfos = new ArrayList<>(infos.length);
-        Collections.addAll(sortedInfos, infos);
-        sortedInfos.sort(new Comparator<>() {
-            @Override
-            public int compare(ThreadInfo o1, ThreadInfo o2) {
-                return o1.getThreadName().toLowerCase().compareTo(o2.getThreadName().toLowerCase());
+    public DiagnosticsDumpElement captureDump(final boolean verbose) {
+        final ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
+        final String threadDump = captureThreadDump(threadMXBean);
+
+        final StringBuilder dumpBuilder = new StringBuilder(threadDump);
+        appendDeadlockedThreadIds(dumpBuilder, "DEADLOCK DETECTED", threadMXBean.findDeadlockedThreads());
+        appendDeadlockedThreadIds(dumpBuilder, "MONITOR DEADLOCK DETECTED", threadMXBean.findMonitorDeadlockedThreads());
+
+        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<ThreadInfo> 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()));
     }
 }

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);
+        }
+    }
+}