From 8083647d34e2efad010d72f60510ee73a379db46 Mon Sep 17 00:00:00 2001 From: sarveshkaushal Date: Wed, 16 Sep 2026 17:51:31 -0700 Subject: [PATCH 1/2] fix: remove Thread.sleep() from DefaultWorkQueueTest --- .../workqueue/DefaultWorkQueueTest.java | 138 +++++++++--------- 1 file changed, 71 insertions(+), 67 deletions(-) diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java index fef4149064..e4ee98ff37 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java @@ -14,22 +14,24 @@ import static org.assertj.core.api.Assertions.assertThat; -import java.util.HashMap; -import java.util.Map; import java.util.concurrent.CountDownLatch; import org.junit.jupiter.api.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; class DefaultWorkQueueTest { - private static final Logger LOGGER = LoggerFactory.getLogger(DefaultWorkQueueTest.class); + private static final long TIMEOUT_SECONDS = 10; @Test void multiProducerAndConsumers() throws Exception { DefaultWorkQueue queue = new DefaultWorkQueue<>(); final int producerCount = 10; final int consumerCount = 5; + Semaphore start = new Semaphore(0); + ConcurrentLinkedQueue failures = new ConcurrentLinkedQueue<>(); // Start producers CountDownLatch producerLatch = new CountDownLatch(producerCount); @@ -39,12 +41,13 @@ void multiProducerAndConsumers() throws Exception { new Thread( () -> { try { + start.acquire(); for (int j = 0; j < 50; j++) { queue.add(String.valueOf(num)); - Thread.sleep(10); } - } catch (Exception e) { - // empty body + } catch (InterruptedException e) { + failures.add(e); + Thread.currentThread().interrupt(); } finally { producerLatch.countDown(); } @@ -55,27 +58,24 @@ void multiProducerAndConsumers() throws Exception { // Start consumers CountDownLatch consumerLatch = new CountDownLatch(consumerCount); for (int i = 0; i < consumerCount; i++) { - final int num = i; Thread t = new Thread( () -> { try { + start.acquire(); for (; ; ) { String item = queue.get(); - assertThat(item) - .withFailMessage("Got an item added after shutdown") - .isNotEqualTo("added after shutdown!"); if (item == null) { return; } - - LOGGER.info("Worker {}: begin processing {}", num, item); - Thread.sleep(50); - LOGGER.info("Worker {}: done processing {}", num, item); + if ("added after shutdown!".equals(item)) { + failures.add(new AssertionError("Got an item added after shutdown")); + } queue.done(item); } - } catch (Exception e) { - // empty body + } catch (InterruptedException e) { + failures.add(e); + Thread.currentThread().interrupt(); } finally { consumerLatch.countDown(); } @@ -83,65 +83,69 @@ void multiProducerAndConsumers() throws Exception { t.start(); } - producerLatch.await(); - queue.shutDown(); + start.release(producerCount + consumerCount); + + boolean producersFinished; + try { + producersFinished = producerLatch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } finally { + queue.shutDown(); + } queue.add("added after shutdown!"); - consumerLatch.await(); + + assertThat(producersFinished).as("producers finished").isTrue(); + assertThat(consumerLatch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) + .as("consumers finished") + .isTrue(); + assertThat(failures).isEmpty(); + assertThat(queue.length()).isZero(); } @Test void addWhileProcessing() throws Exception { DefaultWorkQueue queue = new DefaultWorkQueue<>(); - final int producerCount = 10; - final int consumerCount = 5; + Semaphore itemDequeued = new Semaphore(0); + Semaphore allowDone = new Semaphore(0); + CountDownLatch consumerFinished = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); - // Start producers - CountDownLatch producerLatch = new CountDownLatch(producerCount); - for (int i = 0; i < producerCount; i++) { - final int num = i; - Thread t = - new Thread( - () -> { - queue.add(String.valueOf(num)); - producerLatch.countDown(); - }); - t.start(); - } + queue.add("foo"); - // Start consumers - CountDownLatch consumerLatch = new CountDownLatch(consumerCount); - for (int i = 0; i < consumerCount; i++) { - Thread t = - new Thread( - () -> { - // Every worker will re-add every item up to two times. - // This tests the dirty-while-processing case. - Map counters = new HashMap<>(); - try { - for (; ; ) { - String item = queue.get(); - if (item == null) { - return; - } - counters.putIfAbsent(item, 1); - counters.computeIfPresent(item, (s, integer) -> counters.get(s) + 1); - if (counters.get(item) < 2) { - queue.add(item); - } - queue.done(item); - } - } catch (Exception e) { - // empty body - } finally { - consumerLatch.countDown(); - } - }); - t.start(); + Thread consumer = + new Thread( + () -> { + try { + String item = queue.get(); + itemDequeued.release(); + allowDone.acquire(); + queue.done(item); + } catch (InterruptedException e) { + failure.set(e); + Thread.currentThread().interrupt(); + } finally { + consumerFinished.countDown(); + } + }); + consumer.start(); + + boolean dequeued = itemDequeued.tryAcquire(TIMEOUT_SECONDS, TimeUnit.SECONDS); + try { + assertThat(dequeued).as("consumer dequeued the item").isTrue(); + queue.add("foo"); + } finally { + allowDone.release(); } - producerLatch.await(); - queue.shutDown(); - consumerLatch.await(); + assertThat(consumerFinished.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) + .as("consumer finished") + .isTrue(); + assertThat(failure.get()).isNull(); + assertThat(queue.length()).isEqualTo(1); + + String item = queue.get(); + assertThat(item).isEqualTo("foo"); + queue.done(item); + assertThat(queue.length()).isZero(); } @Test From fa8de8dd33742ba8eef1285e2ddefd58b2e48f84 Mon Sep 17 00:00:00 2001 From: sarveshkaushal Date: Tue, 6 Oct 2026 21:54:21 -0700 Subject: [PATCH 2/2] fix: remove Thread.sleep() from DefaultWorkQueueTest --- .../workqueue/DefaultWorkQueueTest.java | 86 +++++++++++-------- 1 file changed, 48 insertions(+), 38 deletions(-) diff --git a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java index e4ee98ff37..187d82fdfb 100644 --- a/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java +++ b/extended/src/test/java/io/kubernetes/client/extended/workqueue/DefaultWorkQueueTest.java @@ -14,6 +14,8 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.CountDownLatch; import org.junit.jupiter.api.Test; import java.util.concurrent.ConcurrentLinkedQueue; @@ -104,48 +106,56 @@ void multiProducerAndConsumers() throws Exception { @Test void addWhileProcessing() throws Exception { DefaultWorkQueue queue = new DefaultWorkQueue<>(); - Semaphore itemDequeued = new Semaphore(0); - Semaphore allowDone = new Semaphore(0); - CountDownLatch consumerFinished = new CountDownLatch(1); - AtomicReference failure = new AtomicReference<>(); - - queue.add("foo"); + final int producerCount = 10; + final int consumerCount = 5; - Thread consumer = - new Thread( - () -> { - try { - String item = queue.get(); - itemDequeued.release(); - allowDone.acquire(); - queue.done(item); - } catch (InterruptedException e) { - failure.set(e); - Thread.currentThread().interrupt(); - } finally { - consumerFinished.countDown(); - } - }); - consumer.start(); - - boolean dequeued = itemDequeued.tryAcquire(TIMEOUT_SECONDS, TimeUnit.SECONDS); - try { - assertThat(dequeued).as("consumer dequeued the item").isTrue(); - queue.add("foo"); - } finally { - allowDone.release(); + // Start producers + CountDownLatch producerLatch = new CountDownLatch(producerCount); + for (int i = 0; i < producerCount; i++) { + final int num = i; + Thread t = + new Thread( + () -> { + queue.add(String.valueOf(num)); + producerLatch.countDown(); + }); + t.start(); } - assertThat(consumerFinished.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) - .as("consumer finished") - .isTrue(); - assertThat(failure.get()).isNull(); - assertThat(queue.length()).isEqualTo(1); + // Start consumers + CountDownLatch consumerLatch = new CountDownLatch(consumerCount); + for (int i = 0; i < consumerCount; i++) { + Thread t = + new Thread( + () -> { + // Every worker will re-add every item up to two times. + // This tests the dirty-while-processing case. + Map counters = new HashMap<>(); + try { + for (; ; ) { + String item = queue.get(); + if (item == null) { + return; + } + counters.putIfAbsent(item, 1); + counters.computeIfPresent(item, (s, integer) -> counters.get(s) + 1); + if (counters.get(item) < 2) { + queue.add(item); + } + queue.done(item); + } + } catch (Exception e) { + // empty body + } finally { + consumerLatch.countDown(); + } + }); + t.start(); + } - String item = queue.get(); - assertThat(item).isEqualTo("foo"); - queue.done(item); - assertThat(queue.length()).isZero(); + producerLatch.await(); + queue.shutDown(); + consumerLatch.await(); } @Test