Merge pull request #1905 from yue9944882/cherrypick-13-handle-stale-watch-event

Cherrypick-13: Bugfix: stale watch event should be handled properly

Author: k8s-ci-robot

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

@@ -16,6 +16,7 @@
 import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.informer.EventType;
 import io.kubernetes.client.informer.ListerWatcher;
+import io.kubernetes.client.informer.exception.WatchExpiredException;
 import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.openapi.models.V1ListMeta;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
@@ -131,6 +132,12 @@ public void run() {
             watch = newWatch;
           }
           watchHandler(newWatch);
+        } catch (WatchExpiredException e) {
+          // Watch calls were failed due to expired resource-version. Returning
+          // to unwind the list-watch loops so that we can respawn a new round
+          // of list-watching.
+          log.info("{}#Watch expired, will re-list-watch soon", this.apiTypeClass);
+          return;
         } catch (Throwable t) {
           if (isConnectException(t)) {
             // If this is "connection refused" error, it means that most likely
@@ -150,8 +157,10 @@ public void run() {
             continue;
           }
           if ((t instanceof RuntimeException)
+              && t.getMessage() != null
               && t.getMessage().contains("IO Exception during hasNext")) {
             log.info("{}#Read timeout retry list and watch", this.apiTypeClass);
+            // IO timeout should be taken as a normal case
             return;
           }
           this.exceptionHandler.accept(apiTypeClass, t);
@@ -233,7 +242,7 @@ 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());
-          return;
+          throw new WatchExpiredException();
         } else {
           String errorMessage =
               String.format("got ERROR event and its status: %s", item.status.toString());

util/src/main/java/io/kubernetes/client/informer/exception/WatchExpiredException.java

@@ -0,0 +1,16 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package io.kubernetes.client.informer.exception;
+
+/** A ERROR watch item was returned due to stale resource version. */
+public class WatchExpiredException extends RuntimeException {}

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

@@ -14,6 +14,7 @@
 
 import static org.junit.Assert.assertFalse;
 import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -31,6 +32,7 @@
 import io.kubernetes.client.util.Watchable;
 import java.net.HttpURLConnection;
 import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import org.awaitility.Awaitility;
 import org.hamcrest.core.IsEqual;
@@ -279,31 +281,35 @@ public void testReflectorListShouldHandleExpiredResourceVersion() throws ApiExce
     when(listerWatcher.watch(any())).thenThrow(new ApiException(HttpURLConnection.HTTP_GONE, ""));
     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()));
-      assertFalse(reflectorRunnable.isLastSyncResourceVersionUnavailable());
-    } finally {
-      reflectorRunnable.stop();
-    }
+    CompletableFuture<Void> future = CompletableFuture.runAsync(reflectorRunnable::run);
+    Awaitility.await()
+        .atMost(Duration.ofSeconds(2))
+        .pollInterval(Duration.ofMillis(100))
+        .until(() -> future.isDone());
+    assertFalse(future.isCompletedExceptionally());
+  }
 
-    try {
-      when(listerWatcher.list(any())).thenThrow(new ApiException(HttpURLConnection.HTTP_GONE, ""));
-      Thread thread = new Thread(reflectorRunnable::run);
-      thread.setDaemon(true);
-      thread.start();
-      Awaitility.await()
-          .atMost(Duration.ofSeconds(5))
-          .pollInterval(Duration.ofMillis(100))
-          .until(() -> reflectorRunnable.isLastSyncResourceVersionUnavailable());
-    } finally {
-      reflectorRunnable.stop();
-    }
+  @Test
+  public void testReflectorWatchShouldRelistUponExpiredWatchItem() throws ApiException {
+    String expectedResourceVersion = "100";
+    Watchable<V1Pod> expiredWatchable = mock(Watchable.class);
+    when(expiredWatchable.hasNext()).thenReturn(true);
+    when(expiredWatchable.next())
+        .thenReturn(
+            new Watch.Response<>(
+                EventType.ERROR.name(), new V1Status().code(HttpURLConnection.HTTP_GONE)));
+    when(listerWatcher.list(any()))
+        .thenReturn(
+            new V1PodList().metadata(new V1ListMeta().resourceVersion(expectedResourceVersion)));
+    // constantly failing watches will make the reflector run only one time
+    when(listerWatcher.watch(any())).thenReturn(expiredWatchable);
+    ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
+        new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO);
+    CompletableFuture<Void> future = CompletableFuture.runAsync(reflectorRunnable::run);
+    Awaitility.await()
+        .atMost(Duration.ofSeconds(2))
+        .pollInterval(Duration.ofMillis(100))
+        .until(() -> future.isDone());
+    assertFalse(future.isCompletedExceptionally());
   }
 }