fix #3101 making isWatching better reflect health

CHANGELOG.md

@@ -29,6 +29,7 @@
 * Fix #3186: WebSockets and HTTP connections are closed as soon as possible for Watches.
 * Fix #2937: Add `SharedInformerFactory#getExistingSharedIndexInformers` method to return list of registered informers
 * Fix #3239: Add the `Informable` interface for context specific dsl methods to create `SharedIndexInformer`s.
+* Fix #3101: making isWatching a health check for the informer
 
 #### Dependency Upgrade
 * Fix #2741: Update Knative Model to v0.23.0

kubernetes-client/src/main/java/io/fabric8/kubernetes/client/informers/cache/Reflector.java

@@ -81,7 +81,6 @@ private synchronized void stopWatcher() {
    */
   public void listSyncAndWatch() {
     running = true;
-    watching = false;
     final L list = getList();
     final String latestResourceVersion = list.getMetadata().getResourceVersion();
     lastSyncResourceVersion = latestResourceVersion;
@@ -155,12 +154,21 @@ public void eventReceived(Action action, T resource) {
 
     @Override
     public void onClose(WatcherException exception) {
-      watchStopped();
       // this close was triggered by an exception,
       // not the user, it is expected that the watch retry will handle this
       log.warn("Watch closing with exception", exception);
-      if (exception.isHttpGone()) {
+      boolean restarted = false;
+      try {
+        if (exception.isHttpGone()) {
           listSyncAndWatch();
+          restarted = true;
+        } else {
+          running = false; // shouldn't happen, but it means the watch won't restart
+        }
+      } finally {
+        if (!restarted) {
+          watchStopped(); // report the watch as stopped after a problem
+        }
       }
     }
 
@@ -176,5 +184,9 @@ public boolean reconnecting() {
     }
 
   }
+
+  public ReflectorWatcher getWatcher() {
+    return watcher;
+  }
 
 }

kubernetes-client/src/test/java/io/fabric8/kubernetes/client/informers/cache/ReflectorTest.java

@@ -21,6 +21,7 @@
 import io.fabric8.kubernetes.api.model.PodListBuilder;
 import io.fabric8.kubernetes.client.KubernetesClientException;
 import io.fabric8.kubernetes.client.Watch;
+import io.fabric8.kubernetes.client.WatcherException;
 import io.fabric8.kubernetes.client.dsl.base.OperationContext;
 import io.fabric8.kubernetes.client.informers.ListerWatcher;
 import org.junit.jupiter.api.Test;
@@ -29,7 +30,6 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.fail;
 
 class ReflectorTest {
 
@@ -66,5 +66,25 @@ void testStateFlags() {
     assertFalse(reflector.isWatching());
     assertFalse(reflector.isRunning());
   }
+
+  @Test
+  void testNonHttpGone() {
+    ListerWatcher<Pod, PodList> mock = Mockito.mock(ListerWatcher.class);
+    PodList list = new PodListBuilder().withNewMetadata().withResourceVersion("1").endMetadata().build();
+    Mockito.when(mock.list(Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(list);
+
+    Reflector<Pod, PodList> reflector =
+        new Reflector<>(Pod.class, mock, Mockito.mock(SyncableStore.class), new OperationContext());
+
+    reflector.listSyncAndWatch();
+
+    assertTrue(reflector.isWatching());
+    assertTrue(reflector.isRunning());
+
+    reflector.getWatcher().onClose(new WatcherException(null));
+
+    assertFalse(reflector.isWatching());
+    assertFalse(reflector.isRunning());
+  }
 
 }