Fix racy test

Author: brendandburns

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

@@ -25,6 +25,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 
 /** The default delaying queue implementation. */
 public class DefaultDelayingQueue<T> extends DefaultWorkQueue<T> implements DelayingQueue<T> {
@@ -34,8 +35,10 @@ public class DefaultDelayingQueue<T> extends DefaultWorkQueue<T> implements Dela
   private DelayQueue<WaitForEntry<T>> delayQueue;
   private ConcurrentMap<T, WaitForEntry<T>> waitingEntryByData;
   protected BlockingQueue<WaitForEntry<T>> waitingForAddQueue;
+  private Supplier<Instant> timeSource;
 
   public DefaultDelayingQueue(ExecutorService waitingWorker) {
+    this.timeSource = () -> { return Instant.now(); };
     this.delayQueue = new DelayQueue<>();
     this.waitingEntryByData = new ConcurrentHashMap<>();
     this.waitingForAddQueue = new LinkedBlockingQueue<>(1000);
@@ -61,6 +64,11 @@ public void addAfter(T item, Duration duration) {
     this.waitingForAddQueue.offer(entry);
   }
 
+  // Visible for testing
+  protected void injectTimeSource(Supplier<Instant> fn) {
+    this.timeSource = fn;
+  }
+
   private void waitingLoop() {
     try {
       while (true) {

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

@@ -16,6 +16,7 @@
 import io.kubernetes.client.extended.workqueue.ratelimiter.RateLimiter;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.time.Instant;
 
 /** The default rate limiting queue implementation. */
 public class DefaultRateLimitingQueue<T> extends DefaultDelayingQueue<T>

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

@@ -17,19 +17,25 @@
 
 import io.kubernetes.client.extended.wait.Wait;
 import java.time.Duration;
+import java.time.Instant;
 import org.junit.Test;
 
 public class DefaultDelayingQueueTest {
 
   @Test
   public void testSimpleDelayingQueue() throws Exception {
+    final Instant staticTime = Instant.now();
     DefaultDelayingQueue<String> queue = new DefaultDelayingQueue<>();
+    // Hold time still
+    queue.injectTimeSource(() -> { return staticTime; });
     queue.addAfter("foo", Duration.ofMillis(50));
 
+    // Verify that we haven't released it
     assertTrue(waitForWaitingQueueToFill(queue));
     assertTrue(queue.length() == 0);
 
-    Thread.sleep(60L);
+    // Advance time
+    queue.injectTimeSource(() -> { return staticTime.plusMillis(100); });
     assertTrue(waitForAdded(queue, 1));
     String item = queue.get();
     queue.done(item);