Reflector list should handle expired resource version from watch handler as well

Author: aaruna

util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java

@@ -224,7 +224,11 @@ private void watchHandler(Watchable<ApiType> watch) {
       }
       if (eventType.get() == EventType.ERROR) {
         if (item.status != null && item.status.getCode() == HttpURLConnection.HTTP_GONE) {
-          log.info("Watch connection expired: {}", item.status.getMessage());
+          log.info(
+              "ResourceVersion {} and Watch connection expired: {} , will retry w/o resourceVersion next time",
+              getRelistResourceVersion(),
+              item.status.getMessage());
+          isLastSyncResourceVersionUnavailable = true;
           return;
         } else {
           String errorMessage =

util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java

@@ -13,6 +13,7 @@
 package io.kubernetes.client.informer.cache;
 
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -255,4 +256,34 @@ public void testReflectorListShouldHandleExpiredResourceVersion() throws ApiExce
       reflectorRunnable.stop();
     }
   }
+
+  @Test
+  public void testReflectorListShouldHandleExpiredResourceVersionFromWatchHandler()
+      throws ApiException {
+    String expectedResourceVersion = "100";
+    when(listerWatcher.list(any()))
+        .thenReturn(
+            new V1PodList().metadata(new V1ListMeta().resourceVersion(expectedResourceVersion)));
+
+    V1Status v1Status = new V1Status();
+    v1Status.setMessage("dummy-error-message");
+    v1Status.setCode(410);
+    when(listerWatcher.watch(any()))
+        .thenReturn(new MockWatch<>(new Watch.Response("Error", v1Status)));
+    ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
+        new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO);
+    try {
+      Thread thread = new Thread(reflectorRunnable::run);
+      thread.setDaemon(true);
+      thread.start();
+      Awaitility.await()
+          .atMost(Duration.ofSeconds(1))
+          .pollInterval(Duration.ofMillis(100))
+          .until(
+              () -> expectedResourceVersion.equals(reflectorRunnable.getLastSyncResourceVersion()));
+      assertTrue(reflectorRunnable.isLastSyncResourceVersionUnavailable());
+    } finally {
+      reflectorRunnable.stop();
+    }
+  }
 }