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..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,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> queues = new CopyOnWriteArrayList<>(); + + @AfterEach + void tearDown() { + 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( () -> { @@ -53,6 +64,7 @@ void simpleDelayingQueue() throws Exception { void deduping() throws Exception { final Instant staticTime = Instant.now(); DefaultDelayingQueue 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 queue = new DefaultDelayingQueue<>(); + queues.add(queue); 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..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,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 executors = new CopyOnWriteArrayList<>(); + private final List> queues = new CopyOnWriteArrayList<>(); + + @AfterEach + void tearDown() { + queues.forEach(DefaultRateLimitingQueue::shutDown); + executors.forEach(ExecutorService::shutdownNow); + } + private static class MockRateLimiter implements RateLimiter { private int count; @@ -47,9 +60,11 @@ public int numRequeues(Object item) { @Test void simpleRateLimitQueue() throws Exception { - MockRateLimiter mockRateLimiter = new MockRateLimiter<>(); + ExecutorService executor = Executors.newSingleThreadExecutor(); + executors.add(executor); DefaultRateLimitingQueue rlq = - new DefaultRateLimitingQueue<>(Executors.newSingleThreadExecutor(), mockRateLimiter); + new DefaultRateLimitingQueue<>(executor, new MockRateLimiter<>()); + queues.add(rlq); long t1 = System.nanoTime(); rlq.addRateLimited("foo"); rlq.get();