From 5dfc2e8caeea6c77c20c52d0730d59cf4927ec54 Mon Sep 17 00:00:00 2001 From: Brendan Burns <5751682+brendandburns@users.noreply.github.com> Date: Tue, 3 Nov 2026 19:46:56 +0000 Subject: [PATCH 1/4] Initial plan From d78659c15e7fdb0081fb48a33dd30bb4f3a07b98 Mon Sep 17 00:00:00 2001 From: Brendan Burns <5751682+brendandburns@users.noreply.github.com> Date: Tue, 3 Nov 2026 19:50:36 +0000 Subject: [PATCH 2/4] Fix delaying queue executor ownership and shutdown (#4977) --- .../builder/DefaultControllerBuilder.java | 2 +- .../workqueue/DefaultDelayingQueue.java | 26 ++- .../workqueue/DefaultRateLimitingQueue.java | 7 +- .../DefaultDelayingQueueShutdownTest.java | 187 ++++++++++++++++++ .../workqueue/DefaultDelayingQueueTest.java | 11 +- .../workqueue/DefaultRateLimitQueueTest.java | 15 +- 6 files changed, 237 insertions(+), 11 deletions(-) create mode 100644 extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueShutdownTest.java diff --git a/extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java b/extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java index a07fbbe568..a2feb320bb 100644 --- a/extended/src/main/java/io/kubernetes/client/extended/controller/builder/DefaultControllerBuilder.java +++ b/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<>(); } diff --git a/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueue.java b/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueue.java index 835da2b216..5623c6c72e 100644 --- a/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueue.java +++ b/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 extends DefaultWorkQueue implements Dela private ConcurrentMap> waitingEntryByData; protected BlockingQueue> waitingForAddQueue; private Supplier 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) { diff --git a/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java b/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java index 391c75cd22..5947732114 100644 --- a/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java +++ b/extended/src/main/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitingQueue.java @@ -15,7 +15,6 @@ 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 extends DefaultDelayingQueue @@ -23,14 +22,18 @@ public class DefaultRateLimitingQueue extends DefaultDelayingQueue private RateLimiter 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 rateLimiter) { super(waitingWorker); this.rateLimiter = rateLimiter; diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueShutdownTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueShutdownTest.java new file mode 100644 index 0000000000..2ac25eaaa3 --- /dev/null +++ b/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 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 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 unrelated = + executor.submit( + () -> { + started.countDown(); + return release.await(5, TimeUnit.SECONDS); + }); + DefaultDelayingQueue 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 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<>()); + } + } +} diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java index f29791b688..2ec963cd3a 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java @@ -17,14 +17,21 @@ import io.kubernetes.client.extended.wait.Wait; import java.time.Duration; import java.time.Instant; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; class DefaultDelayingQueueTest { + private final DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); + + @AfterEach + void tearDown() { + queue.shutDown(); + } + @Test void simpleDelayingQueue() throws Exception { final Instant staticTime = Instant.now(); - DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); // Hold time still queue.injectTimeSource( () -> { @@ -52,7 +59,6 @@ void simpleDelayingQueue() throws Exception { @Test void deduping() throws Exception { final Instant staticTime = Instant.now(); - DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); String item = "foo"; // Hold time still @@ -111,7 +117,6 @@ void deduping() throws Exception { @Test void copyShifting() throws Exception { final Instant staticTime = Instant.now(); - DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); queue.injectTimeSource( () -> { return staticTime.toEpochMilli(); diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java index 16621d5d44..653d6d405a 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java @@ -16,11 +16,23 @@ import io.kubernetes.client.extended.workqueue.ratelimiter.RateLimiter; import java.time.Duration; +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 ExecutorService executor = Executors.newSingleThreadExecutor(); + private final DefaultRateLimitingQueue rlq = + new DefaultRateLimitingQueue<>(executor, new MockRateLimiter<>()); + + @AfterEach + void tearDown() { + rlq.shutDown(); + executor.shutdownNow(); + } + private static class MockRateLimiter implements RateLimiter { private int count; @@ -47,9 +59,6 @@ public int numRequeues(Object item) { @Test void simpleRateLimitQueue() throws Exception { - MockRateLimiter mockRateLimiter = new MockRateLimiter<>(); - DefaultRateLimitingQueue rlq = - new DefaultRateLimitingQueue<>(Executors.newSingleThreadExecutor(), mockRateLimiter); long t1 = System.nanoTime(); rlq.addRateLimited("foo"); rlq.get(); From 93ba21ebf3e46e8278f1ef7fcc8a2dae637cd5ba Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:44:38 +0000 Subject: [PATCH 3/4] Keep delaying queue tests local and track cleanup separately Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../extended/workqueue/DefaultDelayingQueueTest.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java index 2ec963cd3a..8139091069 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultDelayingQueueTest.java @@ -17,21 +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 DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); + private final List> queues = new CopyOnWriteArrayList<>(); @AfterEach void tearDown() { - queue.shutDown(); + queues.forEach(DefaultDelayingQueue::shutDown); } @Test void simpleDelayingQueue() throws Exception { final Instant staticTime = Instant.now(); + DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); + queues.add(queue); // Hold time still queue.injectTimeSource( () -> { @@ -59,6 +63,8 @@ void simpleDelayingQueue() throws Exception { @Test void deduping() throws Exception { final Instant staticTime = Instant.now(); + DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); + queues.add(queue); String item = "foo"; // Hold time still @@ -117,6 +123,8 @@ void deduping() throws Exception { @Test void copyShifting() throws Exception { final Instant staticTime = Instant.now(); + DefaultDelayingQueue queue = new DefaultDelayingQueue<>(); + queues.add(queue); queue.injectTimeSource( () -> { return staticTime.toEpochMilli(); From 9037d0a167bba70823c3f0ed49e79ad2ba08f3b0 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Nov 2026 20:17:04 +0000 Subject: [PATCH 4/4] Use local rate-limiting queue test resources with tracked teardown Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../workqueue/DefaultRateLimitQueueTest.java | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java index 653d6d405a..9fc9ac541d 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultRateLimitQueueTest.java @@ -16,6 +16,8 @@ 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; @@ -23,14 +25,13 @@ class DefaultRateLimitQueueTest { - private final ExecutorService executor = Executors.newSingleThreadExecutor(); - private final DefaultRateLimitingQueue rlq = - new DefaultRateLimitingQueue<>(executor, new MockRateLimiter<>()); + private final List executors = new CopyOnWriteArrayList<>(); + private final List> queues = new CopyOnWriteArrayList<>(); @AfterEach void tearDown() { - rlq.shutDown(); - executor.shutdownNow(); + queues.forEach(DefaultRateLimitingQueue::shutDown); + executors.forEach(ExecutorService::shutdownNow); } private static class MockRateLimiter implements RateLimiter { @@ -59,6 +60,11 @@ public int numRequeues(Object item) { @Test void simpleRateLimitQueue() throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + executors.add(executor); + DefaultRateLimitingQueue rlq = + new DefaultRateLimitingQueue<>(executor, new MockRateLimiter<>()); + queues.add(rlq); long t1 = System.nanoTime(); rlq.addRateLimited("foo"); rlq.get();