Make sure to close executor services in tests

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/ScheduledExecutorServiceWrapper.java

@@ -128,7 +128,7 @@ public void shutdown() {
 
   @Override
   public List<Runnable> shutdownNow() {
-    this.delegate.shutdownNow();
+    this.scheduler.shutdownNow();
     return this.delegate.shutdownNow();
   }
 

src/test/java/com/rabbitmq/stream/StreamClientTestExtension.java

@@ -0,0 +1,65 @@
+// Copyright (c) 2025 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.stream;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StreamClientTestExtension implements BeforeEachCallback, AfterEachCallback {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(StreamClientTestExtension.class);
+  private static final ExtensionContext.Namespace NS =
+      ExtensionContext.Namespace.create(StreamClientTestExtension.class);
+
+  @Override
+  public void beforeEach(ExtensionContext ctx) {
+    ExtensionContext.Store store = store(ctx);
+    store.put("threads", threads());
+  }
+
+  @Override
+  public void afterEach(ExtensionContext ctx) {
+    @SuppressWarnings("unchecked")
+    Collection<Thread> initialThreads = (Collection<Thread>) store(ctx).remove("threads");
+    if (initialThreads != null) {
+      Collection<Thread> threads = threads();
+      if (threads.size() > initialThreads.size()) {
+        Collection<Thread> diff = new ArrayList<>(threads);
+        diff.removeAll(initialThreads);
+        LOGGER.warn(
+            "[{}] There should be no new threads, initial {}, current {} (diff: {})",
+            ctx.getTestMethod().get().getName(),
+            initialThreads.size(),
+            threads.size(),
+            diff);
+      }
+    } else {
+      LOGGER.warn("No threads in test context");
+    }
+  }
+
+  private static ExtensionContext.Store store(ExtensionContext ctx) {
+    return ctx.getStore(NS);
+  }
+
+  private Collection<Thread> threads() {
+    return Thread.getAllStackTraces().keySet();
+  }
+}

src/test/java/com/rabbitmq/stream/impl/AsyncRetryTest.java

@@ -61,6 +61,7 @@ void callbackCalledIfCompletedImmediately(ScheduledExecutorService scheduler) th
     completableFuture.thenAccept(result::set);
     assertThat(result.get()).isEqualTo(42);
     verify(task, times(1)).call();
+    scheduler.shutdownNow();
   }
 
   @ParameterizedTest
@@ -82,6 +83,7 @@ void shouldRetryWhenExecutionFails(ScheduledExecutorService scheduler) throws Ex
     assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue();
     assertThat(result.get()).isEqualTo(42);
     verify(task, times(3)).call();
+    scheduler.shutdownNow();
   }
 
   @ParameterizedTest
@@ -112,6 +114,7 @@ void shouldTimeoutWhenExecutionFailsForTooLong(ScheduledExecutorService schedule
     assertThat(acceptCalled.get()).isFalse();
     assertThat(exceptionallyCalled.get()).isTrue();
     verify(task, atLeast(5)).call();
+    scheduler.shutdownNow();
   }
 
   @ParameterizedTest
@@ -137,6 +140,7 @@ void shouldRetryWhenPredicateAllowsIt(ScheduledExecutorService scheduler) throws
     assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue();
     assertThat(result.get()).isEqualTo(42);
     verify(task, times(3)).call();
+    scheduler.shutdownNow();
   }
 
   @ParameterizedTest
@@ -171,6 +175,7 @@ void shouldFailWhenPredicateDoesNotAllowRetry(ScheduledExecutorService scheduler
     assertThat(acceptCalled.get()).isFalse();
     assertThat(exceptionallyCalled.get()).isTrue();
     verify(task, times(3)).call();
+    scheduler.shutdownNow();
   }
 
   @ParameterizedTest
@@ -193,6 +198,7 @@ void completeExceptionally(ScheduledExecutorService scheduler) throws Exception
         },
         scheduler);
     assertThat(sync).completes();
+    scheduler.shutdownNow();
   }
 
   static List<ScheduledExecutorService> schedulers() {

src/test/java/com/rabbitmq/stream/impl/ConsumersCoordinatorTest.java

@@ -60,6 +60,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -74,6 +75,7 @@ public class ConsumersCoordinatorTest {
   private static final SubscriptionListener NO_OP_SUBSCRIPTION_LISTENER = subscriptionContext -> {};
   private static final Runnable NO_OP_TRACKING_CLOSING_CALLBACK = () -> {};
 
+  TestInfo info;
   @Mock StreamEnvironment environment;
   @Mock StreamConsumer consumer;
   @Mock Client locator;
@@ -113,7 +115,8 @@ static Stream<Consumer<ConsumersCoordinatorTest>> disruptionArguments() {
   }
 
   @BeforeEach
-  void init() {
+  void init(TestInfo info) {
+    this.info = info;
     Client.ClientParameters clientParameters =
         new Client.ClientParameters() {
           @Override
@@ -169,6 +172,7 @@ public Client.ClientParameters shutdownListener(
 
   @AfterEach
   void tearDown() throws Exception {
+    this.info = null;
     if (coordinator != null) {
       // just taking the opportunity to check toString() generates valid JSON
       MonitoringTestUtils.extract(coordinator);
@@ -2155,15 +2159,16 @@ private MessageListener lastMessageListener() {
     return this.messageListeners.get(messageListeners.size() - 1);
   }
 
-  private static ScheduledExecutorService createScheduledExecutorService() {
+  private ScheduledExecutorService createScheduledExecutorService() {
     return createScheduledExecutorService(1);
   }
 
-  private static ScheduledExecutorService createScheduledExecutorService(int nbThreads) {
+  private ScheduledExecutorService createScheduledExecutorService(int nbThreads) {
+    ThreadFactory tf = ThreadUtils.threadFactory(info.getTestMethod().get().getName() + "-");
     return new ScheduledExecutorServiceWrapper(
         nbThreads == 1
-            ? Executors.newSingleThreadScheduledExecutor()
-            : Executors.newScheduledThreadPool(nbThreads));
+            ? Executors.newSingleThreadScheduledExecutor(tf)
+            : Executors.newScheduledThreadPool(nbThreads, tf));
   }
 
   private static Response responseOk() {

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -59,6 +59,7 @@
 import java.net.ConnectException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -140,11 +141,18 @@ void environmentCreationShouldFailWithUrlUsingWrongPort() throws Exception {
     // no thread leak
     waitAtMost(
         Duration.ofSeconds(20),
-        () -> threads().size() == initialThreads.size(),
-        () ->
-            String.format(
-                "There should be no new threads, initial %s, current %s",
-                initialThreads, threads()));
+        () -> threads().size() <= initialThreads.size(),
+        () -> {
+          Collection<Thread> current = threads();
+          Collection<Thread> diff = Collections.emptySet();
+          if (current.size() > initialThreads.size()) {
+            diff = new ArrayList<>(current);
+            diff.removeAll(initialThreads);
+          }
+          return String.format(
+              "There should be no new threads, initial %d, current %d (diff: %s)",
+              initialThreads.size(), current.size(), diff);
+        });
   }
 
   @Test

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -714,7 +714,6 @@ public void afterAll(ExtensionContext context) {
             try {
               eventLoopGroup.shutdownGracefully(0, 0, SECONDS).get(10, SECONDS);
             } catch (InterruptedException e) {
-              // happens at the end of the test suite
               LOGGER.debug("Error while asynchronously closing Netty event loop group", e);
               Thread.currentThread().interrupt();
             } catch (Exception e) {
@@ -740,7 +739,8 @@ private static class ExecutorServiceCloseableResourceWrapper implements AutoClos
       private final ExecutorService executorService;
 
       private ExecutorServiceCloseableResourceWrapper() {
-        this.executorService = Executors.newCachedThreadPool();
+        ThreadFactory tf = ThreadUtils.threadFactory("closing-resource-");
+        this.executorService = Executors.newCachedThreadPool(tf);
       }
 
       @Override

src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension

@@ -0,0 +1 @@
+com.rabbitmq.stream.StreamClientTestExtension

src/test/resources/junit-platform.properties

@@ -0,0 +1 @@
+junit.jupiter.extensions.autodetection.enabled=false