Clean up watcher test intermittent failures

Author: russgold

operator/src/main/java/oracle/kubernetes/operator/builders/WatchI.java

@@ -3,12 +3,14 @@
 
 package oracle.kubernetes.operator.builders;
 
-import io.kubernetes.client.util.Watch;
-
 import java.util.Iterator;
 
+import io.kubernetes.client.util.Watch;
+
 /**
- * An interface that allows test-stubbing of the Kubernetes Watch class.
+ * An iterator over watch responses from the server. These objects maintain resources,
+ * which will be release when #close() is called.
+ *
  * @param <T> the generic object type
  */
 public interface WatchI<T>

operator/src/test/java/oracle/kubernetes/operator/PodWatcherTest.java

@@ -3,6 +3,12 @@
 
 package oracle.kubernetes.operator;
 
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.collect.ImmutableMap;
+
 import io.kubernetes.client.models.V1ObjectMeta;
 import io.kubernetes.client.models.V1Pod;
 import io.kubernetes.client.models.V1PodCondition;
@@ -14,13 +20,6 @@
 import oracle.kubernetes.operator.work.Packet;
 import oracle.kubernetes.operator.work.Step;
 
-import com.google.common.collect.ImmutableMap;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import org.hamcrest.Matchers;
 import org.junit.Test;
 
@@ -74,7 +73,7 @@ protected <T> T createObjectWithMetaData(V1ObjectMeta metaData) {
 
   @Override
   protected PodWatcher createWatcher(String nameSpace, AtomicBoolean stopping, int initialResourceVersion) {
-    return PodWatcher.create(Executors.defaultThreadFactory(), nameSpace, 
+    return PodWatcher.create(this, nameSpace,
         Integer.toString(initialResourceVersion), this, stopping);
   }
 
@@ -171,7 +170,7 @@ public void whenPodHasServerName_returnIt() throws Exception {
   @Test
   public void waitForReady_returnsAStep() throws Exception {
     AtomicBoolean stopping = new AtomicBoolean(true);
-    PodWatcher watcher = PodWatcher.create(Executors.defaultThreadFactory(), "ns", 
+    PodWatcher watcher = PodWatcher.create(this, "ns",
         Integer.toString(INITIAL_RESOURCE_VERSION), this, stopping);
 
     assertThat(watcher.waitForReady(pod, null), Matchers.instanceOf(Step.class));
@@ -180,7 +179,7 @@ public void waitForReady_returnsAStep() throws Exception {
   @Test
   public void WhenWaitForReadyAppliedToReadyPod_performNextStep() throws Exception {
     AtomicBoolean stopping = new AtomicBoolean(false);
-    PodWatcher watcher = PodWatcher.create(Executors.defaultThreadFactory(), "ns", 
+    PodWatcher watcher = PodWatcher.create(this, "ns",
         Integer.toString(INITIAL_RESOURCE_VERSION), this, stopping);
 
     makePodReady(pod);

operator/src/test/java/oracle/kubernetes/operator/WatcherTestBase.java

@@ -5,6 +5,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.meterware.simplestub.Memento;
@@ -17,7 +18,11 @@
 
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TestRule;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
 
 import static java.net.HttpURLConnection.HTTP_GONE;
 import static oracle.kubernetes.operator.builders.EventMatcher.addEvent;
@@ -30,7 +35,7 @@
 /**
  * Tests behavior of the Watcher class.
  */
-public abstract class WatcherTestBase implements StubWatchFactory.AllWatchesClosedListener {
+public abstract class WatcherTestBase implements StubWatchFactory.AllWatchesClosedListener, ThreadFactory {
     private static final int NEXT_RESOURCE_VERSION = 123456;
     private static final int INITIAL_RESOURCE_VERSION = 123;
     private static final String NAMESPACE = "testspace";
@@ -47,6 +52,25 @@ private V1ObjectMeta createMetaData() {
 
     private AtomicBoolean stopping = new AtomicBoolean(false);
 
+    private String testName;
+    private List<Thread> threads = new ArrayList<>();
+
+    @Rule
+    public TestRule watcher = new TestWatcher() {
+        @Override
+        protected void starting(Description description) {
+            testName = description.getMethodName();
+        }
+    };
+
+    @Override
+    public Thread newThread(Runnable r) {
+        Thread thread = new Thread(r);
+        threads.add(thread);
+        thread.setName(String.format("Test thread %d for %s", threads.size(), testName));
+        return thread;
+    }
+
     @Override
     public void allWatchesClosed() {
         stopping.set(true);
@@ -65,9 +89,18 @@ public void setUp() throws Exception {
 
     @After
     public void tearDown() throws Exception {
+        for (Thread thread : threads) shutDown(thread);
         for (Memento memento : mementos) memento.revert();
     }
 
+    private void shutDown(Thread thread) {
+        try {
+            thread.interrupt();
+            thread.join();
+        } catch (InterruptedException ignored) {
+        }
+    }
+
     @SuppressWarnings("unchecked")
     void sendInitialRequest(int initialResourceVersion) {
         StubWatchFactory.addCallResponses(createAddResponse(createObjectWithMetaData()));

operator/src/test/java/oracle/kubernetes/operator/builders/StubWatchFactory.java

@@ -6,6 +6,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -22,6 +23,8 @@
 import io.kubernetes.client.util.Watch.Response;
 import oracle.kubernetes.operator.helpers.Pool;
 
+import javax.annotation.Nonnull;
+
 /**
  * A test-time replacement for the factory that creates Watch objects, allowing
  * tests to specify directly the events they want returned from the Watch.
@@ -69,15 +72,26 @@ public static List<Map<String, String>> getRecordedParameters() {
     public <T> WatchI<T> createWatch(Pool<ApiClient> pool, CallParams callParams, Class<?> responseBodyType, BiFunction<ApiClient, CallParams, Call> function) throws ApiException {
         getRecordedParameters().add(recordedParams(callParams));
 
-        if (exceptionOnNext == null)
-            return new WatchStub<T>((List)calls.remove(0));
-        else try {
-            return new ExceptionThrowingWatchStub<T>(exceptionOnNext);
-        } finally {
-            exceptionOnNext = null;
+        try {
+            if (nothingToDo())
+                return new WatchStub<>(Collections.emptyList());
+            else if (exceptionOnNext == null)
+                return new WatchStub<T>((List)calls.remove(0));
+            else try {
+                return new ExceptionThrowingWatchStub<>(exceptionOnNext);
+            } finally {
+                exceptionOnNext = null;
+            }
+        } catch (IndexOutOfBoundsException e) {
+            System.out.println("Failed in thread " + Thread.currentThread());
+            throw e;
         }
     }
 
+    public boolean nothingToDo() {
+        return calls.isEmpty() && exceptionOnNext == null;
+    }
+
     private Map<String,String> recordedParams(CallParams callParams) {
         Map<String,String> result = new HashMap<>();
         if (callParams.getResourceVersion() != null)
@@ -116,7 +130,7 @@ public void close() throws IOException {
         }
 
         @Override
-        public Iterator<Watch.Response<T>> iterator() {
+        public @Nonnull Iterator<Watch.Response<T>> iterator() {
             return responses.iterator();
         }