Terminate monitor thread created by the timeout implementation when the supplier returns. Otherwise, with long timeouts, there can be multiple threads in TIMED_WAITING state just wasting memory. New tests for SE and MP. Issue helidon-io#10824. (helidon-io#10833)

Author: spericas

fault-tolerance/fault-tolerance/src/main/java/io/helidon/faulttolerance/TimeoutImpl.java

@@ -18,6 +18,7 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
@@ -93,7 +94,7 @@ public <T> T invoke(Supplier<? extends T> supplier) {
             AtomicBoolean callReturned = new AtomicBoolean(false);
             AtomicBoolean interrupted = new AtomicBoolean(false);
 
-            executor.submit(FaultTolerance.toDelayedRunnable(() -> {
+            Future<?> monitor = executor.submit(FaultTolerance.toDelayedRunnable(() -> {
                 interruptLock.lock();
                 try {
                     if (callReturned.compareAndSet(false, true)) {
@@ -120,6 +121,11 @@ public <T> T invoke(Supplier<? extends T> supplier) {
 
                 interruptLock.lock();
                 try {
+                    boolean cancelMonitor = monitor.cancel(true);
+                    if (!cancelMonitor) {
+                        LOGGER.log(System.Logger.Level.DEBUG, "Unable to cancel monitor thread");
+                    }
+
                     callReturned.set(true);
                     // Run invocation in current thread
                     // Clear interrupted flag here -- required for uninterruptible busy loops

fault-tolerance/fault-tolerance/src/test/java/io/helidon/faulttolerance/TimeoutThreadsTest.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2025 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.faulttolerance;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+
+/**
+ * Check that all threads created by timeout instances are properly
+ * terminated and not in {@link Thread.State#TIMED_WAITING} state.
+ */
+class TimeoutThreadTest {
+
+    @Test
+    void testTimeout() {
+        // create executor with our thread factory
+        TestThreadFactory threadFactory = new TestThreadFactory();
+        ExecutorService executor = Executors.newThreadPerTaskExecutor(threadFactory);
+
+        // run 10 tasks with long timeouts that return immediately
+        for (int i = 0; i < 10; i++) {
+            String status = Timeout.builder()
+                    .timeout(Duration.ofSeconds(60))        // long timeout
+                    .currentThread(true)
+                    .executor(executor)
+                    .build()
+                    .invoke(() -> "done");
+            assertThat(status, is("done"));
+        }
+
+        // if timeout monitor threads stopped, no TIMED_WAITING states
+        List<Thread> threads = threadFactory.threads();
+        assertThat(threads.size(), is(10));
+        for (Thread thread : threads) {
+            assertThat(thread.getState(), is(not(Thread.State.TIMED_WAITING)));
+        }
+    }
+
+    static class TestThreadFactory implements ThreadFactory {
+
+        private final ThreadFactory delegate = Thread.ofVirtual().factory();
+        private final List<Thread> threads = new CopyOnWriteArrayList<>();
+
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = delegate.newThread(r);
+            threads.add(t);
+            return t;
+        }
+
+        List<Thread> threads() {
+            return threads;
+        }
+    }
+}

microprofile/fault-tolerance/src/test/java/io/helidon/microprofile/faulttolerance/TimeoutThreadsTest.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2025 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.microprofile.faulttolerance;
+
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+import io.helidon.faulttolerance.FaultTolerance;
+import io.helidon.microprofile.testing.AddBean;
+import io.helidon.microprofile.testing.junit5.HelidonTest;
+
+import jakarta.inject.Inject;
+import org.eclipse.microprofile.faulttolerance.Timeout;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+
+/**
+ * Check that all threads created by timeout instances are properly
+ * terminated and not in {@link Thread.State#TIMED_WAITING} state.
+ */
+@HelidonTest
+@AddBean(TimeoutThreadsTest.MyBean.class)
+class TimeoutThreadsTest {
+
+    private final static TestThreadFactory THREAD_FACTORY = new TestThreadFactory();
+
+    @Inject
+    private MyBean bean;
+
+    @BeforeAll
+    static void init() {
+        ExecutorService executor = Executors.newThreadPerTaskExecutor(THREAD_FACTORY);
+        FaultTolerance.executor(() -> executor);
+    }
+
+    @Test
+    void testTimeout() {
+        for (int i = 0; i < 10; i++) {
+            String status = bean.timeout();
+            assertThat(status, is("done"));
+        }
+    }
+
+    @AfterAll
+    static void close() {
+        List<Thread> threads = THREAD_FACTORY.threads();
+        assertThat(threads.size(), is(10));
+        for (Thread thread : threads) {
+            assertThat(thread.getState(), is(not(Thread.State.TIMED_WAITING)));
+        }
+    }
+
+    static class MyBean {
+
+        @Timeout(value = 60, unit = ChronoUnit.SECONDS)     // long timeout
+        public String timeout() {
+            return "done";
+        }
+    }
+
+    static class TestThreadFactory implements ThreadFactory {
+
+        private final ThreadFactory delegate = Thread.ofVirtual().factory();
+        private final List<Thread> threads = new CopyOnWriteArrayList<>();
+
+        @Override
+        public Thread newThread(Runnable r) {
+            Thread t = delegate.newThread(r);
+            threads.add(t);
+            return t;
+        }
+
+        List<Thread> threads() {
+            return threads;
+        }
+    }
+}