Thread leaks and thread filters/ exclusions.

Author: dweiss

randomizedtesting-jupiter/src/main/java/com/carrotsearch/randomizedtesting/jupiter/DetectThreadLeaks.java

@@ -6,6 +6,7 @@
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
+import java.util.function.Predicate;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 /**
@@ -35,7 +36,8 @@ enum Scope {
    * Milliseconds to wait for leaked threads to self-terminate before declaring a failure. If all
    * leaked threads terminate within this window, the test passes. Default is 0 (no lingering).
    *
-   * <p>Place this annotation on the same class as {@link DetectThreadLeaks}.
+   * <p>Place this annotation on the same class or method as {@link DetectThreadLeaks}. A
+   * method-level annotation takes precedence over a class-level one.
    */
   @Target({ElementType.TYPE, ElementType.METHOD})
   @Retention(RetentionPolicy.RUNTIME)
@@ -44,4 +46,19 @@ enum Scope {
   @interface LingerTime {
     int millis();
   }
+
+  /**
+   * Excludes threads matched by any of the given {@link Predicate} classes from leak detection. A
+   * thread is excluded when at least one predicate returns {@code true} for it.
+   *
+   * <p>Annotations are collected hierarchically: the test method, then the class, then each
+   * superclass, and the filters from all levels are combined. Place on the same class or method as
+   * {@link DetectThreadLeaks}.
+   */
+  @Target({ElementType.TYPE, ElementType.METHOD})
+  @Retention(RetentionPolicy.RUNTIME)
+  @Documented
+  @interface ExcludeThreads {
+    Class<? extends Predicate<Thread>>[] value();
+  }
 }

randomizedtesting-jupiter/src/main/java/com/carrotsearch/randomizedtesting/jupiter/DetectThreadLeaksExtension.java

@@ -1,8 +1,11 @@
 package com.carrotsearch.randomizedtesting.jupiter;
 
+import java.util.Arrays;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import org.junit.jupiter.api.extension.AfterAllCallback;
@@ -36,7 +39,7 @@ public void beforeAll(ExtensionContext context) {
       return;
     }
     if (scope(context) == DetectThreadLeaks.Scope.SUITE) {
-      context.getStore(EXTENSION_NAMESPACE).put(SNAPSHOT_KEY, liveThreads());
+      context.getStore(EXTENSION_NAMESPACE).put(SNAPSHOT_KEY, liveThreads(buildFilter(context)));
     }
   }
 
@@ -48,15 +51,16 @@ public void afterAll(ExtensionContext context) {
     checkLeaks(
         context.getStore(EXTENSION_NAMESPACE),
         "suite [" + context.getDisplayName() + "]",
-        linger(context));
+        linger(context),
+        buildFilter(context));
   }
 
   @Override
   public void beforeEach(ExtensionContext context) {
     if (isConcurrentMode(context) || scope(context) != DetectThreadLeaks.Scope.TEST) {
       return;
     }
-    context.getStore(EXTENSION_NAMESPACE).put(SNAPSHOT_KEY, liveThreads());
+    context.getStore(EXTENSION_NAMESPACE).put(SNAPSHOT_KEY, liveThreads(buildFilter(context)));
   }
 
   @Override
@@ -67,7 +71,8 @@ public void afterEach(ExtensionContext context) {
     checkLeaks(
         context.getStore(EXTENSION_NAMESPACE),
         "test [" + context.getDisplayName() + "]",
-        linger(context));
+        linger(context),
+        buildFilter(context));
   }
 
   private static DetectThreadLeaks.Scope scope(ExtensionContext context) {
@@ -87,6 +92,51 @@ private static int linger(ExtensionContext context) {
     return classAnn == null ? 0 : classAnn.millis();
   }
 
+  /**
+   * Collects {@link DetectThreadLeaks.ExcludeThreads} filter classes from the entire hierarchy
+   * (method → class → superclasses) and returns a combined predicate that excludes a thread when
+   * any filter matches it.
+   */
+  private static Predicate<Thread> buildFilter(ExtensionContext context) {
+    var filterClasses = new LinkedHashSet<Class<? extends Predicate<Thread>>>();
+
+    context
+        .getTestMethod()
+        .ifPresent(
+            m -> {
+              var ann = m.getAnnotation(DetectThreadLeaks.ExcludeThreads.class);
+              if (ann != null) {
+                for (var c : ann.value()) filterClasses.add(c);
+              }
+            });
+
+    for (Class<?> cls = context.getRequiredTestClass(); cls != null; cls = cls.getSuperclass()) {
+      var ann = cls.getAnnotation(DetectThreadLeaks.ExcludeThreads.class);
+      if (ann != null) {
+        filterClasses.addAll(Arrays.asList(ann.value()));
+      }
+    }
+
+    if (filterClasses.isEmpty()) {
+      return t -> false;
+    }
+
+    var predicates =
+        filterClasses.stream()
+            .map(
+                cls -> {
+                  try {
+                    return (Predicate<Thread>) cls.getDeclaredConstructor().newInstance();
+                  } catch (Exception e) {
+                    throw new RuntimeException(
+                        "Cannot instantiate thread filter: " + cls.getName(), e);
+                  }
+                })
+            .toList();
+
+    return t -> predicates.stream().anyMatch(p -> p.test(t));
+  }
+
   private static boolean isConcurrentMode(ExtensionContext context) {
     // Check the concurrent flag stored in beforeAll (class-level context = parent of method ctx).
     return context
@@ -98,11 +148,12 @@ private static boolean isConcurrentMode(ExtensionContext context) {
         .orElse(false);
   }
 
-  private static void checkLeaks(ExtensionContext.Store store, String description, int lingerMs) {
+  private static void checkLeaks(
+      ExtensionContext.Store store, String description, int lingerMs, Predicate<Thread> filter) {
     var snapshot = store.get(SNAPSHOT_KEY, HashSet.class);
     if (snapshot == null) return;
 
-    var leaked = leakedSince(snapshot);
+    var leaked = leakedSince(snapshot, filter);
     if (leaked.isEmpty()) return;
 
     // Linger: poll until threads self-terminate or the window expires.
@@ -116,7 +167,7 @@ private static void checkLeaks(ExtensionContext.Store store, String description,
           Thread.currentThread().interrupt();
           break;
         }
-        leaked = leakedSince(snapshot);
+        leaked = leakedSince(snapshot, filter);
       }
       if (leaked.isEmpty()) return;
     }
@@ -147,20 +198,22 @@ private static void checkLeaks(ExtensionContext.Store store, String description,
     throw new AssertionError(sb.toString());
   }
 
-  private static Map<Thread, StackTraceElement[]> leakedSince(HashSet<?> snapshot) {
-    var current = liveThreadsWithStacks();
+  private static Map<Thread, StackTraceElement[]> leakedSince(
+      HashSet<?> snapshot, Predicate<Thread> filter) {
+    var current = liveThreadsWithStacks(filter);
     current.keySet().removeAll(snapshot);
     return current;
   }
 
-  private static HashSet<Thread> liveThreads() {
-    return new HashSet<>(liveThreadsWithStacks().keySet());
+  private static HashSet<Thread> liveThreads(Predicate<Thread> filter) {
+    return new HashSet<>(liveThreadsWithStacks(filter).keySet());
   }
 
-  private static Map<Thread, StackTraceElement[]> liveThreadsWithStacks() {
+  private static Map<Thread, StackTraceElement[]> liveThreadsWithStacks(Predicate<Thread> filter) {
     return Thread.getAllStackTraces().entrySet().stream()
         .filter(e -> e.getKey().isAlive())
         .filter(e -> !isKnownSystemThread(e.getKey()))
+        .filter(e -> !filter.test(e.getKey()))
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 

randomizedtesting-jupiter/src/test/java/com/carrotsearch/randomizedtesting/jupiter/F005_ThreadLeaks.java

@@ -249,6 +249,100 @@ void testMethod() {
     }
   }
 
+  @Nested
+  class TestExcludeThreads {
+    @Test
+    void excludedThreadDoesNotFail() {
+      collectExecutionResults(testKitBuilder(ExcludedByClassFilter.class))
+          .results()
+          .allEvents()
+          .assertThatEvents()
+          .doNotHave(event(finishedWithFailure()));
+    }
+
+    @Test
+    void nonExcludedThreadStillFails() {
+      collectExecutionResults(testKitBuilder(NonExcludedStillFails.class))
+          .results()
+          .allEvents()
+          .finished()
+          .failed()
+          .assertEventsMatchExactly(event(finishedWithFailure(instanceOf(AssertionError.class))));
+    }
+
+    @Test
+    void methodAndClassFiltersStackHierarchically() {
+      collectExecutionResults(testKitBuilder(HierarchicalFilters.class))
+          .results()
+          .allEvents()
+          .assertThatEvents()
+          .doNotHave(event(finishedWithFailure()));
+    }
+
+    // Class filter excludes "excluded-a-*"; the leaked thread matches → pass.
+    @DetectThreadLeaks(scope = DetectThreadLeaks.Scope.SUITE)
+    @DetectThreadLeaks.ExcludeThreads(ExcludeNamedAFilter.class)
+    static class ExcludedByClassFilter extends IgnoreInStandaloneRuns {
+      @Test
+      void testMethod() {
+        startNamedThread("excluded-a-1");
+      }
+    }
+
+    // Class filter excludes "excluded-a-*"; leaked thread is unnamed → still detected → fail.
+    @DetectThreadLeaks(scope = DetectThreadLeaks.Scope.SUITE)
+    @DetectThreadLeaks.ExcludeThreads(ExcludeNamedAFilter.class)
+    static class NonExcludedStillFails extends IgnoreInStandaloneRuns {
+      @Test
+      void testMethod() {
+        startSleepingThread();
+      }
+    }
+
+    // Class filter excludes "excluded-a-*", method filter excludes "excluded-b-*";
+    // both threads are started → both excluded → pass.
+    @DetectThreadLeaks(scope = DetectThreadLeaks.Scope.TEST)
+    @DetectThreadLeaks.ExcludeThreads(ExcludeNamedAFilter.class)
+    static class HierarchicalFilters extends IgnoreInStandaloneRuns {
+      @Test
+      @DetectThreadLeaks.ExcludeThreads(ExcludeNamedBFilter.class)
+      void testMethod() {
+        startNamedThread("excluded-a-1");
+        startNamedThread("excluded-b-1");
+      }
+    }
+  }
+
+  /** Predicate that excludes threads whose names start with "excluded-a-". */
+  public static class ExcludeNamedAFilter implements java.util.function.Predicate<Thread> {
+    @Override
+    public boolean test(Thread t) {
+      return t.getName().startsWith("excluded-a-");
+    }
+  }
+
+  /** Predicate that excludes threads whose names start with "excluded-b-". */
+  public static class ExcludeNamedBFilter implements java.util.function.Predicate<Thread> {
+    @Override
+    public boolean test(Thread t) {
+      return t.getName().startsWith("excluded-b-");
+    }
+  }
+
+  private static void startNamedThread(String name) {
+    var t =
+        new Thread(
+            () -> {
+              try {
+                Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+              } catch (InterruptedException ignored) {
+              }
+            },
+            name);
+    t.setDaemon(true);
+    t.start();
+  }
+
   /** Starts a daemon thread that sleeps long enough to be observable as a leak. */
   private static void startSleepingThread() {
     var t =