Terminate owned delaying-queue executors on shutdown

extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java

@@ -46,7 +46,7 @@ public class DefaultControllerBuilder {
   DefaultControllerBuilder() {
     this.workerCount = Constants.DEFAULT_WORKER_COUNT;
     this.controllerName = "default-controller";
-    this.workQueue = new DefaultRateLimitingQueue<>(Executors.newSingleThreadExecutor());
+    this.workQueue = new DefaultRateLimitingQueue<>();
     this.readyFuncs = new ArrayList<>();
   }
 

extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueue.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.Delayed;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -34,21 +35,42 @@ public class DefaultDelayingQueue<T> extends DefaultWorkQueue<T> implements Dela
   private ConcurrentMap<T, WaitForEntry<T>> waitingEntryByData;
   protected BlockingQueue<WaitForEntry<T>> waitingForAddQueue;
   private Supplier<Long> timeSource;
+  private final ExecutorService ownedWaitingWorker;
+  private final Future<?> waitingTask;
 
   private static Long now() {
     return System.nanoTime() / 1000000;
   }
 
+  /**
+   * Creates a queue using a caller-owned executor. Shutdown cancels the waiting task but does not
+   * shut down the executor.
+   */
   public DefaultDelayingQueue(ExecutorService waitingWorker) {
+    this(waitingWorker, false);
+  }
+
+  private DefaultDelayingQueue(ExecutorService waitingWorker, boolean ownsWaitingWorker) {
+    this.ownedWaitingWorker = ownsWaitingWorker ? waitingWorker : null;
     this.timeSource = DefaultDelayingQueue::now;
     this.delayQueue = new DelayQueue<>();
     this.waitingEntryByData = new ConcurrentHashMap<>();
     this.waitingForAddQueue = new LinkedBlockingQueue<>(1000);
-    waitingWorker.submit(this::waitingLoop);
+    this.waitingTask = waitingWorker.submit(this::waitingLoop);
   }
 
+  /** Creates a queue that shuts down its internally owned executor on shutdown. */
   public DefaultDelayingQueue() {
-    this(Executors.newSingleThreadExecutor());
+    this(Executors.newSingleThreadExecutor(), true);
+  }
+
+  @Override
+  public void shutDown() {
+    super.shutDown();
+    waitingTask.cancel(true);
+    if (ownedWaitingWorker != null) {
+      ownedWaitingWorker.shutdown();
+    }
   }
 
   public void addAfter(T item, Duration duration) {

extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java

@@ -15,22 +15,25 @@
 import io.kubernetes.client.extended.workqueue.ratelimiter.DefaultControllerRateLimiter;
 import io.kubernetes.client.extended.workqueue.ratelimiter.RateLimiter;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 /** The default rate limiting queue implementation. */
 public class DefaultRateLimitingQueue<T> extends DefaultDelayingQueue<T>
     implements RateLimitingQueue<T> {
 
   private RateLimiter<T> rateLimiter;
 
+  /** Creates a queue that shuts down its internally owned executor on shutdown. */
   public DefaultRateLimitingQueue() {
-    this(Executors.newSingleThreadExecutor());
+    super();
+    this.rateLimiter = new DefaultControllerRateLimiter<>();
   }
 
+  /** Creates a queue using a caller-owned executor that is not shut down by this queue. */
   public DefaultRateLimitingQueue(ExecutorService waitingWorker) {
     this(waitingWorker, new DefaultControllerRateLimiter<>());
   }
 
+  /** Creates a queue using a caller-owned executor that is not shut down by this queue. */
   public DefaultRateLimitingQueue(ExecutorService waitingWorker, RateLimiter<T> rateLimiter) {
     super(waitingWorker);
     this.rateLimiter = rateLimiter;

extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueShutdownTest.java

@@ -0,0 +1,187 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+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.kubernetes.client.extended.workqueue;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.kubernetes.client.extended.controller.DefaultController;
+import io.kubernetes.client.extended.controller.builder.ControllerBuilder;
+import io.kubernetes.client.extended.controller.builder.DefaultControllerBuilder;
+import io.kubernetes.client.extended.controller.reconciler.Result;
+import io.kubernetes.client.extended.wait.Wait;
+import io.kubernetes.client.extended.workqueue.ratelimiter.DefaultControllerRateLimiter;
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class DefaultDelayingQueueShutdownTest {
+
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void defaultDelayingQueueTerminates(boolean delayed) throws Exception {
+    assertOwnedExecutorTerminates(new DefaultDelayingQueue<>(), delayed);
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  void defaultRateLimitingQueueTerminates(boolean delayed) throws Exception {
+    assertOwnedExecutorTerminates(new DefaultRateLimitingQueue<>(), delayed);
+  }
+
+  private void assertOwnedExecutorTerminates(DefaultDelayingQueue<String> queue, boolean delayed)
+      throws Exception {
+    ExecutorService executor = ownedExecutor(queue);
+    try {
+      if (delayed) {
+        queue.addAfter("later", Duration.ofDays(1));
+        assertThat(
+                Wait.poll(
+                    Duration.ofMillis(10),
+                    Duration.ofSeconds(5),
+                    () -> queue.waitingForAddQueue.isEmpty()))
+            .isTrue();
+      }
+      queue.shutDown();
+      queue.shutDown();
+      assertThat(queue.isShuttingDown()).isTrue();
+      assertThat(executor.isShutdown()).isTrue();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+      queue.addAfter("ignored", Duration.ZERO);
+      assertThat(queue.get()).isNull();
+    } finally {
+      queue.shutDown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    }
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {0, 1, 2})
+  void suppliedExecutorRemainsUsable(int constructor) throws Exception {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    DefaultDelayingQueue<String> queue = suppliedQueue(executor, constructor);
+    try {
+      queue.addAfter("later", Duration.ofDays(1));
+      assertThat(
+              Wait.poll(
+                  Duration.ofMillis(10),
+                  Duration.ofSeconds(5),
+                  () -> queue.waitingForAddQueue.isEmpty()))
+          .isTrue();
+      queue.shutDown();
+      queue.shutDown();
+      assertThat(executor.isShutdown()).isFalse();
+      // On a single-thread executor this can complete only after the waiting loop exits.
+      assertThat(executor.submit(() -> "still usable").get(5, TimeUnit.SECONDS))
+          .isEqualTo("still usable");
+    } finally {
+      queue.shutDown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    }
+  }
+
+  @Test
+  void shutdownBeforeWaitingTaskStartsDoesNotInterruptOtherWork() throws Exception {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    CountDownLatch started = new CountDownLatch(1);
+    CountDownLatch release = new CountDownLatch(1);
+    Future<Boolean> unrelated =
+        executor.submit(
+            () -> {
+              started.countDown();
+              return release.await(5, TimeUnit.SECONDS);
+            });
+    DefaultDelayingQueue<String> queue = new DefaultDelayingQueue<>(executor);
+    try {
+      assertThat(started.await(5, TimeUnit.SECONDS)).isTrue();
+      queue.shutDown();
+      queue.shutDown();
+      Field field = DefaultDelayingQueue.class.getDeclaredField("waitingTask");
+      field.setAccessible(true);
+      assertThat(((Future<?>) field.get(queue)).isCancelled()).isTrue();
+      assertThat(unrelated.isDone()).isFalse();
+      release.countDown();
+      assertThat(unrelated.get(5, TimeUnit.SECONDS)).isTrue();
+      assertThat(executor.submit(() -> "still usable").get(5, TimeUnit.SECONDS))
+          .isEqualTo("still usable");
+    } finally {
+      release.countDown();
+      queue.shutDown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    }
+  }
+
+  @Test
+  void controllerShutdownTerminatesQueueExecutor() throws Exception {
+    DefaultController controller =
+        (DefaultController)
+            ControllerBuilder.defaultBuilder(null)
+                .withReconciler(request -> new Result(false))
+                .build();
+    ExecutorService executor =
+        ownedExecutor((DefaultDelayingQueue<?>) controller.getWorkQueue());
+    try {
+      controller.shutdown();
+      controller.shutdown();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    } finally {
+      controller.shutdown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    }
+  }
+
+  @Test
+  void replacingControllerQueueTerminatesOriginalExecutor() throws Exception {
+    DefaultControllerBuilder builder = ControllerBuilder.defaultBuilder(null);
+    Field field = DefaultControllerBuilder.class.getDeclaredField("workQueue");
+    field.setAccessible(true);
+    DefaultDelayingQueue<?> queue = (DefaultDelayingQueue<?>) field.get(builder);
+    ExecutorService executor = ownedExecutor(queue);
+    try {
+      builder.withWorkQueue(null);
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    } finally {
+      queue.shutDown();
+      executor.shutdownNow();
+      assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+    }
+  }
+
+  private static ExecutorService ownedExecutor(DefaultDelayingQueue<?> queue) throws Exception {
+    Field field = DefaultDelayingQueue.class.getDeclaredField("ownedWaitingWorker");
+    field.setAccessible(true);
+    return (ExecutorService) field.get(queue);
+  }
+
+  private static DefaultDelayingQueue<String> suppliedQueue(
+      ExecutorService executor, int constructor) {
+    switch (constructor) {
+      case 0:
+        return new DefaultDelayingQueue<>(executor);
+      case 1:
+        return new DefaultRateLimitingQueue<>(executor);
+      default:
+        return new DefaultRateLimitingQueue<>(executor, new DefaultControllerRateLimiter<>());
+    }
+  }
+}

extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java

@@ -17,14 +17,25 @@
 import io.kubernetes.client.extended.wait.Wait;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
 class DefaultDelayingQueueTest {
 
+  private final List<DefaultDelayingQueue<String>> queues = new CopyOnWriteArrayList<>();
+
+  @AfterEach
+  void tearDown() {
+    queues.forEach(DefaultDelayingQueue::shutDown);
+  }
+
   @Test
   void simpleDelayingQueue() throws Exception {
     final Instant staticTime = Instant.now();
     DefaultDelayingQueue<String> queue = new DefaultDelayingQueue<>();
+    queues.add(queue);
     // Hold time still
     queue.injectTimeSource(
         () -> {
@@ -53,6 +64,7 @@ void simpleDelayingQueue() throws Exception {
   void deduping() throws Exception {
     final Instant staticTime = Instant.now();
     DefaultDelayingQueue<String> queue = new DefaultDelayingQueue<>();
+    queues.add(queue);
     String item = "foo";
 
     // Hold time still
@@ -112,6 +124,7 @@ void deduping() throws Exception {
   void copyShifting() throws Exception {
     final Instant staticTime = Instant.now();
     DefaultDelayingQueue<String> queue = new DefaultDelayingQueue<>();
+    queues.add(queue);
     queue.injectTimeSource(
         () -> {
           return staticTime.toEpochMilli();

extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java

@@ -16,11 +16,24 @@
 
 import io.kubernetes.client.extended.workqueue.ratelimiter.RateLimiter;
 import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
 class DefaultRateLimitQueueTest {
 
+  private final List<ExecutorService> executors = new CopyOnWriteArrayList<>();
+  private final List<DefaultRateLimitingQueue<String>> queues = new CopyOnWriteArrayList<>();
+
+  @AfterEach
+  void tearDown() {
+    queues.forEach(DefaultRateLimitingQueue::shutDown);
+    executors.forEach(ExecutorService::shutdownNow);
+  }
+
   private static class MockRateLimiter<T> implements RateLimiter<T> {
 
     private int count;
@@ -47,9 +60,11 @@ public int numRequeues(Object item) {
 
   @Test
   void simpleRateLimitQueue() throws Exception {
-    MockRateLimiter<String> mockRateLimiter = new MockRateLimiter<>();
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    executors.add(executor);
     DefaultRateLimitingQueue<String> rlq =
-        new DefaultRateLimitingQueue<>(Executors.newSingleThreadExecutor(), mockRateLimiter);
+        new DefaultRateLimitingQueue<>(executor, new MockRateLimiter<>());
+    queues.add(rlq);
     long t1 = System.nanoTime();
     rlq.addRateLimited("foo");
     rlq.get();