From 9077a067962131282476a04d8f13a672985b8a5f Mon Sep 17 00:00:00 2001 From: Brendan Burns <5751682+brendandburns@users.noreply.github.com> Date: Tue, 1 Sep 2026 19:19:49 +0000 Subject: [PATCH] Add exponential backoff for informer watch reconnects Retry transient informer watch connection failures with bounded exponential backoff, interruption handling, and coverage. --- .../e2e/informer/NamespaceInformerTest.java | 76 ++++++++++++++++++ .../informer/cache/ReflectorRunnable.java | 46 ++++++++++- .../informer/cache/ReflectorRunnableTest.java | 78 +++++++++++++++++++ 3 files changed, 196 insertions(+), 4 deletions(-) diff --git a/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java b/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java index bbf92916b2..4b8647d6eb 100644 --- a/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java +++ b/e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java @@ -15,14 +15,25 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import io.kubernetes.client.informer.ListerWatcher; +import io.kubernetes.client.informer.ResourceEventHandler; import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.SharedInformerFactory; import io.kubernetes.client.informer.cache.Lister; import io.kubernetes.client.openapi.ApiClient; +import io.kubernetes.client.openapi.ApiException; +import io.kubernetes.client.openapi.apis.CoreV1Api; import io.kubernetes.client.openapi.models.V1Namespace; import io.kubernetes.client.openapi.models.V1NamespaceList; +import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.util.CallGeneratorParams; import io.kubernetes.client.util.ClientBuilder; +import io.kubernetes.client.util.Watchable; import io.kubernetes.client.util.generic.GenericKubernetesApi; +import io.kubernetes.client.util.generic.options.ListOptions; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; class NamespaceInformerTest { @@ -57,4 +68,69 @@ void listWatchingNamespaces() throws Exception { informerFactory.stopAllRegisteredInformers(true); } } + + @Test + void listWatchingNamespacesRecoversFromInitialConnectExceptions() throws Exception { + ApiClient client = ClientBuilder.defaultClient(); + CoreV1Api coreV1Api = new CoreV1Api(client); + SharedInformerFactory informerFactory = new SharedInformerFactory(client); + String namespaceName = "e2e-informer-retry"; + AtomicInteger watchAttempts = new AtomicInteger(0); + GenericKubernetesApi api = + new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client); + + ListerWatcher flakyWatcher = + new ListerWatcher() { + @Override + public V1NamespaceList list(CallGeneratorParams params) { + return api + .list( + new ListOptions() + .resourceVersion(params.resourceVersion) + .timeoutSeconds(params.timeoutSeconds)) + .getObject(); + } + + @Override + public Watchable watch(CallGeneratorParams params) throws ApiException { + if (watchAttempts.incrementAndGet() <= 2) { + throw new RuntimeException(new java.net.ConnectException("simulated transient failure")); + } + return api.watch( + new ListOptions() + .resourceVersion(params.resourceVersion) + .timeoutSeconds(params.timeoutSeconds)); + } + }; + + SharedIndexInformer nsInformer = + informerFactory.sharedIndexInformerFor(flakyWatcher, V1Namespace.class, 0); + CountDownLatch selectedSeen = new CountDownLatch(1); + try { + nsInformer.addEventHandler( + new ResourceEventHandler() { + @Override + public void onAdd(V1Namespace obj) { + if (namespaceName.equals(obj.getMetadata().getName())) { + selectedSeen.countDown(); + } + } + + @Override + public void onUpdate(V1Namespace oldObj, V1Namespace newObj) {} + + @Override + public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {} + }); + + informerFactory.startAllRegisteredInformers(); + await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue()); + coreV1Api.createNamespace(new V1Namespace().metadata(new V1ObjectMeta().name(namespaceName))).execute(); + assertThat(selectedSeen.await(45, TimeUnit.SECONDS)).isTrue(); + assertThat(watchAttempts.get()).isGreaterThanOrEqualTo(3); + } finally { + informerFactory.stopAllRegisteredInformers(true); + coreV1Api.deleteNamespace(namespaceName).execute(); + } + } } diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java b/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java index d77e5afab7..b4ed27119a 100644 --- a/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java +++ b/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java @@ -34,6 +34,7 @@ import java.util.Optional; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; +import java.util.function.LongConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +47,8 @@ public class ReflectorRunnable< public static Duration REFLECTOR_WATCH_CLIENTSIDE_MAX_TIMEOUT = Duration.ofMinutes(5 * 2); private static final Logger log = LoggerFactory.getLogger(ReflectorRunnable.class); + private static final long WATCH_RETRY_INITIAL_BACKOFF_MILLIS = 1000L; + private static final long WATCH_RETRY_MAX_BACKOFF_MILLIS = 30000L; private String lastSyncResourceVersion; @@ -65,6 +68,8 @@ public class ReflectorRunnable< private Method setKindMethod; private Method setApiVersionMethod; + private final LongConsumer connectExceptionSleeper; + private long watchRetryBackoffMillis; public ReflectorRunnable( Class apiTypeClass, @@ -78,11 +83,22 @@ public ReflectorRunnable( ListerWatcher listerWatcher, DeltaFIFO store, BiConsumer, Throwable> exceptionHandler) { + this(apiTypeClass, listerWatcher, store, exceptionHandler, ReflectorRunnable::sleep); + } + + ReflectorRunnable( + Class apiTypeClass, + ListerWatcher listerWatcher, + DeltaFIFO store, + BiConsumer, Throwable> exceptionHandler, + LongConsumer connectExceptionSleeper) { this.listerWatcher = listerWatcher; this.store = store; this.apiTypeClass = apiTypeClass; this.exceptionHandler = exceptionHandler == null ? ReflectorRunnable::defaultWatchErrorHandler : exceptionHandler; + this.connectExceptionSleeper = connectExceptionSleeper; + this.watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS; try { this.setKindMethod = apiTypeClass.getMethod("setKind", String.class); this.setApiVersionMethod = apiTypeClass.getMethod("setApiVersion", String.class); @@ -97,6 +113,9 @@ public ReflectorRunnable( */ public void run() { log.info("{}#Start listing and watching...", apiTypeClass); + // run() can be invoked multiple times for the same reflector instance; always restart backoff + // from the initial value for each list-watch cycle. + resetWatchRetryBackoff(); try { ApiListType list = @@ -148,6 +167,7 @@ public void run() { watch = newWatch; } watchHandler(newWatch); + resetWatchRetryBackoff(); } 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 @@ -161,10 +181,9 @@ public void run() { // objects because most likely we will be able to restart watch where // we ended. If that's the case wait and resend watch request. log.info("{}#Watch get connect exception, retry watch", this.apiTypeClass); - try { - Thread.sleep(1000L); - } catch (InterruptedException e) { - // no-op + sleepForConnectExceptionRetry(); + if (Thread.currentThread().isInterrupted()) { + return; } continue; } @@ -364,4 +383,23 @@ private boolean isConnectException(Throwable t) { Throwable cause = t.getCause(); return cause instanceof ConnectException; } + + private void sleepForConnectExceptionRetry() { + long currentBackoffMillis = watchRetryBackoffMillis; + watchRetryBackoffMillis = + Math.min(watchRetryBackoffMillis * 2, WATCH_RETRY_MAX_BACKOFF_MILLIS); + connectExceptionSleeper.accept(currentBackoffMillis); + } + + private void resetWatchRetryBackoff() { + watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS; + } + + private static void sleep(long durationMillis) { + try { + Thread.sleep(durationMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } } diff --git a/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java b/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java index d4c81a26ca..01b45fbd54 100644 --- a/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java +++ b/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java @@ -33,11 +33,13 @@ import io.kubernetes.client.util.Watchable; import java.net.HttpURLConnection; import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; import org.awaitility.Awaitility; @@ -361,6 +363,82 @@ void reflectorListShouldHandleExpiredResourceVersionFromWatchHandler() } } + @Test + void reflectorWatchConnectExceptionShouldUseExponentialBackoff() + throws ApiException, InterruptedException { + List retryBackoffs = new ArrayList<>(); + CountDownLatch latch = new CountDownLatch(3); + when(listerWatcher.list(any())) + .thenReturn(new V1PodList().metadata(new V1ListMeta().resourceVersion("100"))); + when(listerWatcher.watch(any())).thenThrow(new RuntimeException(new java.net.ConnectException("refused"))); + ReflectorRunnable reflectorRunnable = + new ReflectorRunnable<>( + V1Pod.class, + listerWatcher, + deltaFIFO, + exceptionHandler, + backoff -> { + if (retryBackoffs.size() < 3) { + retryBackoffs.add(backoff); + latch.countDown(); + } + }); + try { + Thread thread = new Thread(reflectorRunnable::run); + thread.setDaemon(true); + thread.start(); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + } finally { + reflectorRunnable.stop(); + } + assertThat(retryBackoffs).containsExactly(1000L, 2000L, 4000L); + } + + @Test + void reflectorWatchBackoffShouldResetAfterSuccessfulWatch() { + List retryBackoffs = new ArrayList<>(); + CountDownLatch latch = new CountDownLatch(2); + AtomicInteger watchCount = new AtomicInteger(); + + ReflectorRunnable reflectorRunnable = + new ReflectorRunnable<>( + V1Pod.class, + new ListerWatcher() { + @Override + public V1PodList list(CallGeneratorParams params) { + return new V1PodList().metadata(new V1ListMeta().resourceVersion("100")); + } + + @Override + public Watchable watch(CallGeneratorParams params) { + int call = watchCount.incrementAndGet(); + if (call == 2) { + return new MockWatch<>(); + } + throw new RuntimeException(new java.net.ConnectException("refused")); + } + }, + deltaFIFO, + exceptionHandler, + backoff -> { + if (retryBackoffs.size() < 2) { + retryBackoffs.add(backoff); + latch.countDown(); + } + }); + try { + Thread thread = new Thread(reflectorRunnable::run); + thread.setDaemon(true); + thread.start(); + assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + reflectorRunnable.stop(); + } + assertThat(retryBackoffs).containsExactly(1000L, 1000L); + } + @Test void defaultExceptionHandlerSetPerDefault() { ReflectorRunnable reflector =