Recover informer relisting from ResourceVersionTooLarge errors

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

@@ -19,8 +19,10 @@
 import io.kubernetes.client.informer.ListerWatcher;
 import io.kubernetes.client.informer.exception.WatchExpiredException;
 import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.JSON;
 import io.kubernetes.client.openapi.models.V1ListMeta;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Status;
 import io.kubernetes.client.util.CallGeneratorParams;
 import io.kubernetes.client.util.ModelMapper;
 import io.kubernetes.client.util.Strings;
@@ -182,7 +184,7 @@ public void run() {
         }
       }
     } catch (ApiException e) {
-      if (e.getCode() == HttpURLConnection.HTTP_GONE) {
+      if (e.getCode() == HttpURLConnection.HTTP_GONE || isTooLargeResourceVersion(e)) {
         log.info(
             "ResourceVersion {} expired, will retry w/o resourceVersion at the next time",
             getRelistResourceVersion());
@@ -195,6 +197,24 @@ public void run() {
     }
   }
 
+  private static boolean isTooLargeResourceVersion(ApiException exception) {
+    if (Strings.isNullOrEmpty(exception.getResponseBody())) {
+      return false;
+    }
+    V1Status status;
+    try {
+      status = JSON.deserialize(exception.getResponseBody(), V1Status.class);
+    } catch (RuntimeException e) {
+      // Generated status adapters can throw unchecked exceptions for malformed response bodies.
+      return false;
+    }
+    return status != null
+        && status.getDetails() != null
+        && status.getDetails().getCauses() != null
+        && status.getDetails().getCauses().stream()
+            .anyMatch(cause -> cause != null && "ResourceVersionTooLarge".equals(cause.getReason()));
+  }
+
   public void stop() {
     try {
       isActive.set(false);

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

@@ -43,6 +43,10 @@
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
@@ -212,6 +216,113 @@ void reflectorRunnableShouldCaptureListNon410ApiException() throws ApiException
     }
   }
 
+  @ParameterizedTest
+  @ValueSource(ints = {410, 504})
+  void reflectorRecoversUnavailableResourceVersion(int statusCode) throws ApiException {
+    V1Pod initialPod = new V1Pod().metadata(new V1ObjectMeta().name("initial"));
+    V1Pod recoveredPod = new V1Pod().metadata(new V1ObjectMeta().name("recovered"));
+    ApiException unavailable =
+        statusCode == 410
+            ? new ApiException(410, "Gone")
+            : new ApiException(
+                statusCode,
+                null,
+                "{\"kind\":\"Status\",\"apiVersion\":\"v1\",\"reason\":\"Timeout\","
+                    + "\"details\":{\"causes\":[{\"reason\":\"Other\"},"
+                    + "{\"reason\":\"ResourceVersionTooLarge\"}]},\"code\":504}");
+    when(listerWatcher.list(any()))
+        .thenReturn(
+            new V1PodList()
+                .metadata(new V1ListMeta().resourceVersion("100"))
+                .items(Arrays.asList(initialPod)))
+        .thenThrow(unavailable)
+        .thenReturn(
+            new V1PodList()
+                .metadata(new V1ListMeta().resourceVersion("200"))
+                .items(Arrays.asList(recoveredPod)));
+    MockWatch<V1Pod> initialWatch =
+        new MockWatch<>(new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500)));
+    MockWatch<V1Pod> recoveredWatch =
+        new MockWatch<>(
+            new Watch.Response<>(
+                EventType.BOOKMARK.name(),
+                new V1Pod().metadata(new V1ObjectMeta().resourceVersion("201"))),
+            new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500)));
+    when(listerWatcher.watch(any())).thenReturn(initialWatch, recoveredWatch);
+    ReflectorRunnable<V1Pod, V1PodList> reflector =
+        new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO, exceptionHandler);
+
+    reflector.run();
+    assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100");
+    reflector.run();
+    assertThat(reflector.isLastSyncResourceVersionUnavailable()).isTrue();
+    assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100");
+    verify(exceptionHandler, never()).accept(V1Pod.class, unavailable);
+    reflector.run();
+
+    ArgumentCaptor<CallGeneratorParams> listParams =
+        ArgumentCaptor.forClass(CallGeneratorParams.class);
+    verify(listerWatcher, times(3)).list(listParams.capture());
+    assertThat(listParams.getAllValues())
+        .extracting(params -> params.resourceVersion)
+        .containsExactly("0", "100", "");
+    ArgumentCaptor<CallGeneratorParams> watchParams =
+        ArgumentCaptor.forClass(CallGeneratorParams.class);
+    verify(listerWatcher, times(2)).watch(watchParams.capture());
+    assertThat(watchParams.getAllValues())
+        .extracting(params -> params.resourceVersion)
+        .containsExactly("100", "200");
+    verify(deltaFIFO).replace(Arrays.asList(initialPod), "100");
+    verify(deltaFIFO).replace(Arrays.asList(recoveredPod), "200");
+    assertThat(reflector.isLastSyncResourceVersionUnavailable()).isFalse();
+    assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("201");
+    assertThat(initialWatch.isClosed()).isTrue();
+    assertThat(recoveredWatch.isClosed()).isTrue();
+  }
+
+  @ParameterizedTest
+  @NullAndEmptySource
+  @ValueSource(
+      strings = {
+        "not json",
+        "null",
+        "[]",
+        "{}",
+        "{\"details\":null}",
+        "{\"details\":{\"causes\":null}}",
+        "{\"details\":{\"causes\":[]}}",
+        "{\"details\":{\"causes\":[null,{}]}}",
+        "{\"details\":{\"causes\":\"invalid\"}}",
+        "{\"details\":{\"causes\":[{\"reason\":{}}]}}",
+        "{\"reason\":\"Timeout\",\"message\":\"Too large resource version\"}",
+        "{\"reason\":\"Timeout\",\"details\":{\"causes\":[{\"reason\":\"Other\"}]}}"
+      })
+  void reflectorReportsOrdinaryTimeoutsAndMalformedBodies(String body) throws ApiException {
+    ApiException timeout = new ApiException(504, null, body);
+    when(listerWatcher.list(any()))
+        .thenReturn(new V1PodList().metadata(new V1ListMeta().resourceVersion("100")))
+        .thenThrow(timeout);
+    when(listerWatcher.watch(any()))
+        .thenReturn(
+            new MockWatch<>(
+                new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500))));
+    ReflectorRunnable<V1Pod, V1PodList> reflector =
+        new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO, exceptionHandler);
+
+    reflector.run();
+    reflector.run();
+    reflector.run();
+
+    ArgumentCaptor<CallGeneratorParams> params = ArgumentCaptor.forClass(CallGeneratorParams.class);
+    verify(listerWatcher, times(3)).list(params.capture());
+    assertThat(params.getAllValues())
+        .extracting(value -> value.resourceVersion)
+        .containsExactly("0", "100", "100");
+    verify(exceptionHandler, times(2)).accept(V1Pod.class, timeout);
+    assertThat(reflector.isLastSyncResourceVersionUnavailable()).isFalse();
+    assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100");
+  }
+
   @Test
   void reflectorRunnableCaptureWatchException() throws ApiException {
     RuntimeException expectedException = new RuntimeException("noxu");