From 899cf7a70a7d88bdf70f2a98e0b952f68669c5cf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:12:37 +0000 Subject: [PATCH 1/8] Add ListOptions watch parity fields for GenericKubernetesApi Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../client/e2e/dynamic/DynamicApiTest.java | 33 ++++++++++++ .../util/generic/GenericKubernetesApi.java | 8 +++ .../util/generic/options/ListOptions.java | 32 ++++++++++++ .../generic/GenericKubernetesApiTest.java | 50 +++++++++++++++++++ 4 files changed, 123 insertions(+) diff --git a/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java b/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java index 3c76ff7a03..5949d71914 100644 --- a/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java +++ b/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java @@ -19,7 +19,10 @@ import io.kubernetes.client.openapi.models.V1Namespace; import io.kubernetes.client.openapi.models.V1ObjectMeta; import io.kubernetes.client.util.ClientBuilder; +import io.kubernetes.client.util.generic.KubernetesApiResponse; +import io.kubernetes.client.util.generic.options.ListOptions; import io.kubernetes.client.util.generic.dynamic.DynamicKubernetesApi; +import io.kubernetes.client.util.generic.dynamic.DynamicKubernetesListObject; import io.kubernetes.client.util.generic.dynamic.DynamicKubernetesObject; import io.kubernetes.client.util.generic.dynamic.Dynamics; import org.junit.jupiter.api.Test; @@ -41,4 +44,34 @@ void dynamicApiCreateAndDeleteNamespace() throws Exception { dynamicApi.delete("e2e-dynamic").throwsApiException().getObject(); assertThat(deleted).isNotNull(); } + + @Test + void dynamicApiListWithResourceVersionMatchAndWatchBookmarks() throws Exception { + ApiClient client = ClientBuilder.defaultClient(); + DynamicKubernetesApi dynamicApi = + new DynamicKubernetesApi("", "v1", "namespaces", client); + String namespaceName = "e2e-dynamic-list-options"; + V1Namespace namespace = new V1Namespace().metadata(new V1ObjectMeta().name(namespaceName)); + + DynamicKubernetesObject createdNamespace = + dynamicApi.create(Dynamics.newFromJson(JSON.serialize(namespace))).getObject(); + assertThat(createdNamespace).isNotNull(); + + try { + KubernetesApiResponse listResponse = + dynamicApi.list( + new ListOptions() + .fieldSelector("metadata.name=" + namespaceName) + .resourceVersion(createdNamespace.getMetadata().getResourceVersion()) + .resourceVersionMatch("NotOlderThan") + .allowWatchBookmarks(true)); + + assertThat(listResponse.isSuccess()).isTrue(); + assertThat(listResponse.getObject()).isNotNull(); + assertThat(listResponse.getObject().getItems()) + .anySatisfy(item -> assertThat(item.getMetadata().getName()).isEqualTo(namespaceName)); + } finally { + dynamicApi.delete(namespaceName).throwsApiException(); + } + } } diff --git a/util/src/main/java/io/kubernetes/client/util/generic/GenericKubernetesApi.java b/util/src/main/java/io/kubernetes/client/util/generic/GenericKubernetesApi.java index 21360a7bb0..53893b937a 100644 --- a/util/src/main/java/io/kubernetes/client/util/generic/GenericKubernetesApi.java +++ b/util/src/main/java/io/kubernetes/client/util/generic/GenericKubernetesApi.java @@ -579,11 +579,13 @@ private CallBuilder makeClusterListCallBuilder(final ListOptions listOptions) { adaptListCall( customObjectsApi.getApiClient(), customObjectsApi.listClusterCustomObject( this.apiGroup, this.apiVersion, this.resourcePlural) + .allowWatchBookmarks(listOptions.getAllowWatchBookmarks()) ._continue(listOptions.getContinue()) .fieldSelector(listOptions.getFieldSelector()) .labelSelector(listOptions.getLabelSelector()) .limit(listOptions.getLimit()) .resourceVersion(listOptions.getResourceVersion()) + .resourceVersionMatch(listOptions.getResourceVersionMatch()) .timeoutSeconds(listOptions.getTimeoutSeconds()) .watch(false) .buildCall(null), @@ -626,11 +628,13 @@ private CallBuilder makeNamespacedListCallBuilder( adaptListCall( customObjectsApi.getApiClient(), customObjectsApi.listNamespacedCustomObject(this.apiGroup, this.apiVersion, namespace, this.resourcePlural) + .allowWatchBookmarks(listOptions.getAllowWatchBookmarks()) ._continue(listOptions.getContinue()) .fieldSelector(listOptions.getFieldSelector()) .labelSelector(listOptions.getLabelSelector()) .limit(listOptions.getLimit()) .resourceVersion(listOptions.getResourceVersion()) + .resourceVersionMatch(listOptions.getResourceVersionMatch()) .timeoutSeconds(listOptions.getTimeoutSeconds()) .watch(false) .buildCall(null), @@ -1229,11 +1233,13 @@ public Watchable watch(final ListOptions listOptions) throws ApiExcepti this.apiGroup, this.apiVersion, this.resourcePlural) + .allowWatchBookmarks(listOptions.getAllowWatchBookmarks()) ._continue(listOptions.getContinue()) .fieldSelector(listOptions.getFieldSelector()) .labelSelector(listOptions.getLabelSelector()) .limit(listOptions.getLimit()) .resourceVersion(listOptions.getResourceVersion()) + .resourceVersionMatch(listOptions.getResourceVersionMatch()) .timeoutSeconds(listOptions.getTimeoutSeconds()) .watch(true) .buildCall(null); @@ -1264,11 +1270,13 @@ public Watchable watch(String namespace, final ListOptions listOptions) this.apiVersion, namespace, this.resourcePlural) + .allowWatchBookmarks(listOptions.getAllowWatchBookmarks()) ._continue(listOptions.getContinue()) .fieldSelector(listOptions.getFieldSelector()) .labelSelector(listOptions.getLabelSelector()) .limit(listOptions.getLimit()) .resourceVersion(listOptions.getResourceVersion()) + .resourceVersionMatch(listOptions.getResourceVersionMatch()) .timeoutSeconds(listOptions.getTimeoutSeconds()) .watch(true) .buildCall(null); diff --git a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java index aa386573b8..a2c79082cd 100644 --- a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java +++ b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java @@ -24,6 +24,12 @@ public class ListOptions { @SerializedName("resourceVersion") private String resourceVersion; + @SerializedName("resourceVersionMatch") + private String resourceVersionMatch; + + @SerializedName("allowWatchBookmarks") + private Boolean allowWatchBookmarks; + @SerializedName("timeoutSeconds") private Integer timeoutSeconds; @@ -75,6 +81,32 @@ public void setResourceVersion(String resourceVersion) { this.resourceVersion = resourceVersion; } + public ListOptions resourceVersionMatch(String resourceVersionMatch) { + this.resourceVersionMatch = resourceVersionMatch; + return this; + } + + public String getResourceVersionMatch() { + return resourceVersionMatch; + } + + public void setResourceVersionMatch(String resourceVersionMatch) { + this.resourceVersionMatch = resourceVersionMatch; + } + + public ListOptions allowWatchBookmarks(Boolean allowWatchBookmarks) { + this.allowWatchBookmarks = allowWatchBookmarks; + return this; + } + + public Boolean getAllowWatchBookmarks() { + return allowWatchBookmarks; + } + + public void setAllowWatchBookmarks(Boolean allowWatchBookmarks) { + this.allowWatchBookmarks = allowWatchBookmarks; + } + public ListOptions limit(Integer limit) { this.limit = limit; return this; diff --git a/util/src/test/java/io/kubernetes/client/util/generic/GenericKubernetesApiTest.java b/util/src/test/java/io/kubernetes/client/util/generic/GenericKubernetesApiTest.java index d423cf1a88..b262a03c91 100644 --- a/util/src/test/java/io/kubernetes/client/util/generic/GenericKubernetesApiTest.java +++ b/util/src/test/java/io/kubernetes/client/util/generic/GenericKubernetesApiTest.java @@ -115,6 +115,32 @@ void listNamespacedJobReturningObject() { apiServer.verify(1, getRequestedFor(urlPathEqualTo("/apis/batch/v1/namespaces/default/jobs"))); } + @Test + void listNamespacedJobWithResourceVersionMatchAndWatchBookmarks() { + V1JobList jobList = new V1JobList().kind("JobList").metadata(new V1ListMeta()); + + apiServer.stubFor( + get(urlPathEqualTo("/apis/batch/v1/namespaces/default/jobs")) + .willReturn(aResponse().withStatus(200).withBody(json.serialize(jobList)))); + + KubernetesApiResponse jobListResp = + jobClient.list( + "default", + new ListOptions() + .resourceVersion("123") + .resourceVersionMatch("NotOlderThan") + .allowWatchBookmarks(true)); + assertThat(jobListResp.isSuccess()).isTrue(); + assertThat(jobListResp.getObject()).isEqualTo(jobList); + assertThat(jobListResp.getStatus()).isNull(); + apiServer.verify( + 1, + getRequestedFor(urlPathEqualTo("/apis/batch/v1/namespaces/default/jobs")) + .withQueryParam("resourceVersion", equalTo("123")) + .withQueryParam("resourceVersionMatch", equalTo("NotOlderThan")) + .withQueryParam("allowWatchBookmarks", equalTo("true"))); + } + @Test void listNamespacedJobWithPartialMetadataObjectListHeader() { V1JobList jobList = @@ -279,6 +305,30 @@ void watchNamespacedJobReturningObject() throws ApiException { .withQueryParam("watch", equalTo("true"))); } + @Test + void watchNamespacedJobWithResourceVersionMatchAndWatchBookmarks() throws ApiException { + V1JobList jobList = new V1JobList().kind("JobList").metadata(new V1ListMeta()); + + apiServer.stubFor( + get(urlPathEqualTo("/apis/batch/v1/namespaces/default/jobs")) + .willReturn(aResponse().withStatus(200).withBody(json.serialize(jobList)))); + Watchable jobListWatch = + jobClient.watch( + "default", + new ListOptions() + .resourceVersion("123") + .resourceVersionMatch("NotOlderThan") + .allowWatchBookmarks(true)); + assertThat((Object) jobListWatch).isNotNull(); + apiServer.verify( + 1, + getRequestedFor(urlPathEqualTo("/apis/batch/v1/namespaces/default/jobs")) + .withQueryParam("watch", equalTo("true")) + .withQueryParam("resourceVersion", equalTo("123")) + .withQueryParam("resourceVersionMatch", equalTo("NotOlderThan")) + .withQueryParam("allowWatchBookmarks", equalTo("true"))); + } + @Test void readTimeoutShouldThrowException() { ApiClient apiClient = From 133b4345ce8bad6485d5a0da04c66224af5cd6a6 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:22:00 +0000 Subject: [PATCH 2/8] Add predicate-aware informer event handlers Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../e2e/informer/NamespaceInformerTest.java | 61 ++++++++++++ .../FilteringResourceEventHandler.java | 64 +++++++++++++ .../client/informer/SharedInformer.java | 20 ++++ .../impl/DefaultSharedIndexInformer.java | 21 +++- .../FilteringResourceEventHandlerTest.java | 96 +++++++++++++++++++ 5 files changed, 260 insertions(+), 2 deletions(-) create mode 100644 util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java create mode 100644 util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java 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..42213167bf 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 @@ -17,12 +17,18 @@ import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.SharedInformerFactory; +import io.kubernetes.client.informer.ResourceEventHandler; import io.kubernetes.client.informer.cache.Lister; import io.kubernetes.client.openapi.ApiClient; +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.ClientBuilder; import io.kubernetes.client.util.generic.GenericKubernetesApi; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.junit.jupiter.api.Test; class NamespaceInformerTest { @@ -57,4 +63,59 @@ void listWatchingNamespaces() throws Exception { informerFactory.stopAllRegisteredInformers(true); } } + + @Test + void listWatchingNamespacesWithPredicateHandler() throws Exception { + ApiClient client = ClientBuilder.defaultClient(); + CoreV1Api coreV1Api = new CoreV1Api(client); + SharedInformerFactory informerFactory = new SharedInformerFactory(client); + String selectedNamespace = "e2e-filtered-selected"; + String ignoredNamespace = "e2e-filtered-ignored"; + + coreV1Api.createNamespace(new V1Namespace().metadata(new V1ObjectMeta().name(selectedNamespace))).execute(); + coreV1Api.createNamespace(new V1Namespace().metadata(new V1ObjectMeta().name(ignoredNamespace))).execute(); + + GenericKubernetesApi api = + new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client); + + SharedIndexInformer nsInformer = + informerFactory.sharedIndexInformerFor(api, V1Namespace.class, 0); + CountDownLatch selectedSeen = new CountDownLatch(1); + AtomicBoolean ignoredSeen = new AtomicBoolean(false); + AtomicBoolean selectedSeenByHandler = new AtomicBoolean(false); + try { + nsInformer.addEventHandler( + new ResourceEventHandler() { + @Override + public void onAdd(V1Namespace obj) { + String name = obj.getMetadata().getName(); + if (selectedNamespace.equals(name)) { + selectedSeenByHandler.set(true); + selectedSeen.countDown(); + } + if (ignoredNamespace.equals(name)) { + ignoredSeen.set(true); + } + } + + @Override + public void onUpdate(V1Namespace oldObj, V1Namespace newObj) {} + + @Override + public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {} + }, + ns -> selectedNamespace.equals(ns.getMetadata().getName())); + + informerFactory.startAllRegisteredInformers(); + + await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue()); + assertThat(selectedSeen.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(selectedSeenByHandler).isTrue(); + assertThat(ignoredSeen).isFalse(); + } finally { + informerFactory.stopAllRegisteredInformers(true); + coreV1Api.deleteNamespace(selectedNamespace).execute(); + coreV1Api.deleteNamespace(ignoredNamespace).execute(); + } + } } diff --git a/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java b/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java new file mode 100644 index 0000000000..6ae53efb62 --- /dev/null +++ b/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java @@ -0,0 +1,64 @@ +/* +Copyright 2026 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; + +import io.kubernetes.client.common.KubernetesObject; +import java.util.Objects; +import java.util.function.Predicate; + +/** + * FilteringResourceEventHandler dispatches events only when objects match a predicate. + */ +public class FilteringResourceEventHandler + implements ResourceEventHandler { + + private final ResourceEventHandler delegate; + private final Predicate filter; + + public FilteringResourceEventHandler( + ResourceEventHandler delegate, Predicate filter) { + this.delegate = Objects.requireNonNull(delegate); + this.filter = Objects.requireNonNull(filter); + } + + @Override + public void onAdd(ApiType obj) { + if (filter.test(obj)) { + delegate.onAdd(obj); + } + } + + @Override + public void onUpdate(ApiType oldObj, ApiType newObj) { + boolean oldMatched = oldObj != null && filter.test(oldObj); + boolean newMatched = newObj != null && filter.test(newObj); + if (oldMatched && newMatched) { + delegate.onUpdate(oldObj, newObj); + return; + } + if (!oldMatched && newMatched) { + delegate.onAdd(newObj); + return; + } + if (oldMatched) { + delegate.onDelete(oldObj, false); + } + } + + @Override + public void onDelete(ApiType obj, boolean deletedFinalStateUnknown) { + if (filter.test(obj)) { + delegate.onDelete(obj, deletedFinalStateUnknown); + } + } +} diff --git a/util/src/main/java/io/kubernetes/client/informer/SharedInformer.java b/util/src/main/java/io/kubernetes/client/informer/SharedInformer.java index 8c656fa2b7..2a7a4c3a51 100644 --- a/util/src/main/java/io/kubernetes/client/informer/SharedInformer.java +++ b/util/src/main/java/io/kubernetes/client/informer/SharedInformer.java @@ -13,6 +13,7 @@ package io.kubernetes.client.informer; import io.kubernetes.client.common.KubernetesObject; +import java.util.function.Predicate; /* * SharedInformer defines basic methods of a informer. @@ -26,6 +27,14 @@ public interface SharedInformer { */ void addEventHandler(ResourceEventHandler handler); + /** + * Add event handler with predicate filter. + * + * @param handler the handler + * @param filter the object filter + */ + void addEventHandler(ResourceEventHandler handler, Predicate filter); + /** * addEventHandlerWithResyncPeriod adds an event handler to the shared informer using the * specified resync period. Events to a single handler are delivered sequentially, but there is no @@ -36,6 +45,17 @@ public interface SharedInformer { */ void addEventHandlerWithResyncPeriod(ResourceEventHandler handler, long resyncPeriod); + /** + * addEventHandlerWithResyncPeriod adds an event handler with the specified resync period and + * object filter. + * + * @param handler the event handler + * @param resyncPeriod the specific resync period + * @param filter the object filter + */ + void addEventHandlerWithResyncPeriod( + ResourceEventHandler handler, long resyncPeriod, Predicate filter); + /** run starts the shared informer, which will be stopped until stop() is called. */ void run(); diff --git a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java index 4fd1db93a3..7c1690b66f 100644 --- a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java +++ b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java @@ -18,6 +18,7 @@ import io.kubernetes.client.informer.ResourceEventHandler; import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.TransformFunc; +import io.kubernetes.client.informer.FilteringResourceEventHandler; import io.kubernetes.client.informer.cache.Cache; import io.kubernetes.client.informer.cache.Controller; import io.kubernetes.client.informer.cache.DeltaFIFO; @@ -31,6 +32,7 @@ import java.util.concurrent.ThreadFactory; import java.util.function.BiConsumer; import java.util.function.Function; +import java.util.function.Predicate; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.tuple.MutablePair; import org.slf4j.Logger; @@ -152,13 +154,27 @@ public DefaultSharedIndexInformer( /** add event callback */ @Override public void addEventHandler(ResourceEventHandler handler) { - addEventHandlerWithResyncPeriod(handler, defaultEventHandlerResyncPeriod); + addEventHandler(handler, obj -> true); + } + + @Override + public void addEventHandler( + ResourceEventHandler handler, Predicate filterPredicate) { + addEventHandlerWithResyncPeriod(handler, defaultEventHandlerResyncPeriod, filterPredicate); } /** add event callback with a resync period */ @Override public void addEventHandlerWithResyncPeriod( ResourceEventHandler handler, long resyncPeriodMillis) { + addEventHandlerWithResyncPeriod(handler, resyncPeriodMillis, obj -> true); + } + + @Override + public void addEventHandlerWithResyncPeriod( + ResourceEventHandler handler, + long resyncPeriodMillis, + Predicate filterPredicate) { if (stopped) { log.info( "DefaultSharedIndexInformer#Handler was not added to shared informer because it has stopped already"); @@ -195,7 +211,8 @@ public void addEventHandlerWithResyncPeriod( ProcessorListener listener = new ProcessorListener( - handler, determineResyncPeriod(resyncCheckPeriodMillis, this.resyncCheckPeriodMillis)); + new FilteringResourceEventHandler<>(handler, filterPredicate), + determineResyncPeriod(resyncCheckPeriodMillis, this.resyncCheckPeriodMillis)); if (!started) { this.processor.addListener(listener); return; diff --git a/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java b/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java new file mode 100644 index 0000000000..5f04e3e663 --- /dev/null +++ b/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java @@ -0,0 +1,96 @@ +/* +Copyright 2026 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; + +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.openapi.models.V1Pod; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class FilteringResourceEventHandlerTest { + + private static V1Pod pod(String name) { + return new V1Pod().metadata(new V1ObjectMeta().namespace("default").name(name)); + } + + @Test + void dropsAddForFilteredOutObject() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + + handler.onAdd(pod("ignored")); + + verifyNoInteractions(delegate); + } + + @Test + void convertsUpdateTransitionIntoAdd() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + V1Pod oldObj = pod("ignored"); + V1Pod newObj = pod("selected"); + + handler.onUpdate(oldObj, newObj); + + verify(delegate).onAdd(newObj); + verify(delegate, never()).onUpdate(oldObj, newObj); + verify(delegate, never()).onDelete(oldObj, false); + } + + @Test + void convertsUpdateTransitionIntoDelete() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + V1Pod oldObj = pod("selected"); + V1Pod newObj = pod("ignored"); + + handler.onUpdate(oldObj, newObj); + + verify(delegate).onDelete(oldObj, false); + verify(delegate, never()).onUpdate(oldObj, newObj); + verify(delegate, never()).onAdd(newObj); + } + + @Test + void preservesUpdateWhenOldAndNewMatch() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + V1Pod oldObj = pod("selected"); + V1Pod newObj = pod("selected"); + + handler.onUpdate(oldObj, newObj); + + verify(delegate).onUpdate(oldObj, newObj); + verify(delegate, never()).onAdd(newObj); + verify(delegate, never()).onDelete(oldObj, false); + } + + @Test + void dropsDeleteForFilteredOutObject() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + + handler.onDelete(pod("ignored"), false); + + verifyNoInteractions(delegate); + } +} From bdf553bd52f04d586daa86866af71dd36deaf526 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:26:17 +0000 Subject: [PATCH 3/8] Add exponential backoff for informer watch reconnects Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../e2e/informer/NamespaceInformerTest.java | 71 ++++++++++++++++++ .../informer/cache/ReflectorRunnable.java | 42 +++++++++-- .../informer/cache/ReflectorRunnableTest.java | 74 +++++++++++++++++++ 3 files changed, 182 insertions(+), 5 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 42213167bf..0a00bb30c3 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 @@ -18,17 +18,23 @@ import io.kubernetes.client.informer.SharedIndexInformer; import io.kubernetes.client.informer.SharedInformerFactory; import io.kubernetes.client.informer.ResourceEventHandler; +import io.kubernetes.client.informer.ListerWatcher; 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.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; class NamespaceInformerTest { @@ -118,4 +124,69 @@ public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {} coreV1Api.deleteNamespace(ignoredNamespace).execute(); } } + + @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..4c8ea2a5cc 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); @@ -147,6 +163,7 @@ public void run() { } watch = newWatch; } + resetWatchRetryBackoff(); watchHandler(newWatch); } catch (WatchExpiredException e) { // Watch calls were failed due to expired resource-version. Returning @@ -161,11 +178,7 @@ 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(); continue; } if ((t instanceof RuntimeException) @@ -364,4 +377,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..24619f6464 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,78 @@ 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 -> { + retryBackoffs.add(backoff); + latch.countDown(); + }); + try { + Thread thread = new Thread(reflectorRunnable::run); + thread.setDaemon(true); + thread.start(); + assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); + } finally { + reflectorRunnable.stop(); + } + assertThat(retryBackoffs).startsWith(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 -> { + 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 = From 39d866fd5cac99b97ced2026bd06d911e4d7999f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:28:15 +0000 Subject: [PATCH 4/8] Add fluent builder for generic ListOptions Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../client/e2e/dynamic/DynamicApiTest.java | 5 +- .../util/generic/options/ListOptions.java | 61 +++++++++++++++++++ .../options/ListOptionsBuilderTest.java | 46 ++++++++++++++ 3 files changed, 110 insertions(+), 2 deletions(-) create mode 100644 util/src/test/java/io/kubernetes/client/util/generic/options/ListOptionsBuilderTest.java diff --git a/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java b/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java index 5949d71914..aac5a314ef 100644 --- a/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java +++ b/e2e/src/test/java/io/kubernetes/client/e2e/dynamic/DynamicApiTest.java @@ -60,11 +60,12 @@ void dynamicApiListWithResourceVersionMatchAndWatchBookmarks() throws Exception try { KubernetesApiResponse listResponse = dynamicApi.list( - new ListOptions() + ListOptions.builder() .fieldSelector("metadata.name=" + namespaceName) .resourceVersion(createdNamespace.getMetadata().getResourceVersion()) .resourceVersionMatch("NotOlderThan") - .allowWatchBookmarks(true)); + .allowWatchBookmarks(true) + .build()); assertThat(listResponse.isSuccess()).isTrue(); assertThat(listResponse.getObject()).isNotNull(); diff --git a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java index a2c79082cd..539200c7c0 100644 --- a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java +++ b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java @@ -42,6 +42,10 @@ public class ListOptions { @SerializedName("isPartialObjectMetadataListRequest") private Boolean isPartialObjectMetadataListRequest; + public static Builder builder() { + return new Builder(); + } + public ListOptions fieldSelector(String fieldSelector) { this.fieldSelector = fieldSelector; return this; @@ -159,4 +163,61 @@ public Boolean isPartialObjectMetadataListRequest() { public void setPartialObjectMetadataListRequest(Boolean isPartialObjectMetadataListRequest) { this.isPartialObjectMetadataListRequest = isPartialObjectMetadataListRequest; } + + public static final class Builder { + private final ListOptions options; + + private Builder() { + this.options = new ListOptions(); + } + + public Builder fieldSelector(String fieldSelector) { + options.fieldSelector(fieldSelector); + return this; + } + + public Builder labelSelector(String labelSelector) { + options.labelSelector(labelSelector); + return this; + } + + public Builder resourceVersion(String resourceVersion) { + options.resourceVersion(resourceVersion); + return this; + } + + public Builder resourceVersionMatch(String resourceVersionMatch) { + options.resourceVersionMatch(resourceVersionMatch); + return this; + } + + public Builder allowWatchBookmarks(Boolean allowWatchBookmarks) { + options.allowWatchBookmarks(allowWatchBookmarks); + return this; + } + + public Builder timeoutSeconds(Integer timeoutSeconds) { + options.timeoutSeconds(timeoutSeconds); + return this; + } + + public Builder limit(Integer limit) { + options.limit(limit); + return this; + } + + public Builder _continue(String _continue) { + options._continue(_continue); + return this; + } + + public Builder isPartialObjectMetadataListRequest(Boolean partialMetadata) { + options.isPartialObjectMetadataListRequest(partialMetadata); + return this; + } + + public ListOptions build() { + return options; + } + } } diff --git a/util/src/test/java/io/kubernetes/client/util/generic/options/ListOptionsBuilderTest.java b/util/src/test/java/io/kubernetes/client/util/generic/options/ListOptionsBuilderTest.java new file mode 100644 index 0000000000..9a2f5f012d --- /dev/null +++ b/util/src/test/java/io/kubernetes/client/util/generic/options/ListOptionsBuilderTest.java @@ -0,0 +1,46 @@ +/* +Copyright 2026 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.util.generic.options; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +class ListOptionsBuilderTest { + + @Test + void listOptionsBuilderShouldPopulateAllFields() { + ListOptions options = + ListOptions.builder() + .fieldSelector("metadata.name=test") + .labelSelector("app=test") + .resourceVersion("100") + .resourceVersionMatch("NotOlderThan") + .allowWatchBookmarks(true) + .timeoutSeconds(30) + .limit(50) + ._continue("next-page-token") + .isPartialObjectMetadataListRequest(true) + .build(); + + assertThat(options.getFieldSelector()).isEqualTo("metadata.name=test"); + assertThat(options.getLabelSelector()).isEqualTo("app=test"); + assertThat(options.getResourceVersion()).isEqualTo("100"); + assertThat(options.getResourceVersionMatch()).isEqualTo("NotOlderThan"); + assertThat(options.getAllowWatchBookmarks()).isTrue(); + assertThat(options.getTimeoutSeconds()).isEqualTo(30); + assertThat(options.getLimit()).isEqualTo(50); + assertThat(options.getContinue()).isEqualTo("next-page-token"); + assertThat(options.isPartialObjectMetadataListRequest()).isTrue(); + } +} From 6821c7f2bd14fac08d0dea94dd74374e61458679 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:32:33 +0000 Subject: [PATCH 5/8] Address review feedback and harden backoff behavior Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../client/e2e/informer/NamespaceInformerTest.java | 4 ++-- .../client/informer/cache/ReflectorRunnable.java | 6 +++++- .../client/util/generic/options/ListOptions.java | 12 +++++++++++- .../informer/cache/ReflectorRunnableTest.java | 14 +++++++++----- 4 files changed, 27 insertions(+), 9 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 0a00bb30c3..4b5e79b60f 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 @@ -116,8 +116,8 @@ public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {} await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue()); assertThat(selectedSeen.await(30, TimeUnit.SECONDS)).isTrue(); - assertThat(selectedSeenByHandler).isTrue(); - assertThat(ignoredSeen).isFalse(); + assertThat(selectedSeenByHandler.get()).isTrue(); + assertThat(ignoredSeen.get()).isFalse(); } finally { informerFactory.stopAllRegisteredInformers(true); coreV1Api.deleteNamespace(selectedNamespace).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 4c8ea2a5cc..02e397a68f 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 @@ -113,6 +113,7 @@ public ReflectorRunnable( */ public void run() { log.info("{}#Start listing and watching...", apiTypeClass); + resetWatchRetryBackoff(); try { ApiListType list = @@ -163,8 +164,8 @@ public void run() { } watch = newWatch; } - resetWatchRetryBackoff(); 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 @@ -179,6 +180,9 @@ public void run() { // we ended. If that's the case wait and resend watch request. log.info("{}#Watch get connect exception, retry watch", this.apiTypeClass); sleepForConnectExceptionRetry(); + if (Thread.currentThread().isInterrupted()) { + return; + } continue; } if ((t instanceof RuntimeException) diff --git a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java index 539200c7c0..4bab93b0e1 100644 --- a/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java +++ b/util/src/main/java/io/kubernetes/client/util/generic/options/ListOptions.java @@ -217,7 +217,17 @@ public Builder isPartialObjectMetadataListRequest(Boolean partialMetadata) { } public ListOptions build() { - return options; + ListOptions built = new ListOptions(); + built.setFieldSelector(options.getFieldSelector()); + built.setLabelSelector(options.getLabelSelector()); + built.setResourceVersion(options.getResourceVersion()); + built.setResourceVersionMatch(options.getResourceVersionMatch()); + built.setAllowWatchBookmarks(options.getAllowWatchBookmarks()); + built.setTimeoutSeconds(options.getTimeoutSeconds()); + built.setLimit(options.getLimit()); + built.setContinue(options.getContinue()); + built.setPartialObjectMetadataListRequest(options.isPartialObjectMetadataListRequest()); + return built; } } } 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 24619f6464..73ff478fa9 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 @@ -378,8 +378,10 @@ void reflectorWatchConnectExceptionShouldUseExponentialBackoff() deltaFIFO, exceptionHandler, backoff -> { - retryBackoffs.add(backoff); - latch.countDown(); + if (retryBackoffs.size() < 3) { + retryBackoffs.add(backoff); + latch.countDown(); + } }); try { Thread thread = new Thread(reflectorRunnable::run); @@ -389,7 +391,7 @@ void reflectorWatchConnectExceptionShouldUseExponentialBackoff() } finally { reflectorRunnable.stop(); } - assertThat(retryBackoffs).startsWith(1000L, 2000L, 4000L); + assertThat(retryBackoffs).containsExactly(1000L, 2000L, 4000L); } @Test @@ -419,8 +421,10 @@ public Watchable watch(CallGeneratorParams params) { deltaFIFO, exceptionHandler, backoff -> { - retryBackoffs.add(backoff); - latch.countDown(); + if (retryBackoffs.size() < 2) { + retryBackoffs.add(backoff); + latch.countDown(); + } }); try { Thread thread = new Thread(reflectorRunnable::run); From 0aac39e186edafdf3e9ee588e365becdf8373d02 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:34:23 +0000 Subject: [PATCH 6/8] Harden predicate filtering for null or malformed objects Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../FilteringResourceEventHandler.java | 19 +++++++++++++++---- .../FilteringResourceEventHandlerTest.java | 11 +++++++++++ 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java b/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java index 6ae53efb62..c0bf4f2a73 100644 --- a/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java +++ b/util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java @@ -33,15 +33,15 @@ public FilteringResourceEventHandler( @Override public void onAdd(ApiType obj) { - if (filter.test(obj)) { + if (matches(obj)) { delegate.onAdd(obj); } } @Override public void onUpdate(ApiType oldObj, ApiType newObj) { - boolean oldMatched = oldObj != null && filter.test(oldObj); - boolean newMatched = newObj != null && filter.test(newObj); + boolean oldMatched = matches(oldObj); + boolean newMatched = matches(newObj); if (oldMatched && newMatched) { delegate.onUpdate(oldObj, newObj); return; @@ -57,8 +57,19 @@ public void onUpdate(ApiType oldObj, ApiType newObj) { @Override public void onDelete(ApiType obj, boolean deletedFinalStateUnknown) { - if (filter.test(obj)) { + if (matches(obj)) { delegate.onDelete(obj, deletedFinalStateUnknown); } } + + private boolean matches(ApiType obj) { + if (obj == null) { + return false; + } + try { + return filter.test(obj); + } catch (RuntimeException e) { + return false; + } + } } diff --git a/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java b/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java index 5f04e3e663..3a6a78ba33 100644 --- a/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java +++ b/util/src/test/java/io/kubernetes/client/informer/FilteringResourceEventHandlerTest.java @@ -93,4 +93,15 @@ void dropsDeleteForFilteredOutObject() { verifyNoInteractions(delegate); } + + @Test + void shouldIgnoreDeleteWhenPredicateThrows() { + ResourceEventHandler delegate = Mockito.mock(ResourceEventHandler.class); + FilteringResourceEventHandler handler = + new FilteringResourceEventHandler<>(delegate, obj -> "selected".equals(obj.getMetadata().getName())); + + handler.onDelete(new V1Pod(), true); + + verifyNoInteractions(delegate); + } } From cc3386401df45641177cb6e7d162fa74da2d57a8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:35:48 +0000 Subject: [PATCH 7/8] Document run-cycle backoff reset intent and reduce test flake Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../io/kubernetes/client/informer/cache/ReflectorRunnable.java | 2 ++ .../kubernetes/client/informer/cache/ReflectorRunnableTest.java | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) 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 02e397a68f..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 @@ -113,6 +113,8 @@ 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 { 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 73ff478fa9..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 @@ -387,7 +387,7 @@ void reflectorWatchConnectExceptionShouldUseExponentialBackoff() Thread thread = new Thread(reflectorRunnable::run); thread.setDaemon(true); thread.start(); - assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); } finally { reflectorRunnable.stop(); } From dd9e2a7d57c7a5cda6ee7296b31092c6726ff3d5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 1 Sep 2026 16:49:43 +0000 Subject: [PATCH 8/8] Add cache-level predicate support for informers Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com> --- .../e2e/informer/NamespaceInformerTest.java | 58 +++++++++++++++ .../informer/SharedInformerFactory.java | 74 ++++++++++++++++++- .../client/informer/cache/Cache.java | 36 ++++++++- .../impl/DefaultSharedIndexInformer.java | 14 +++- .../client/informer/cache/CacheTest.java | 49 ++++++++++++ ...SharedIndexInformerCachePredicateTest.java | 64 ++++++++++++++++ 6 files changed, 288 insertions(+), 7 deletions(-) create mode 100644 util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java 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 4b5e79b60f..c2c4fb5210 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 @@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; class NamespaceInformerTest { @@ -189,4 +190,61 @@ public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {} coreV1Api.deleteNamespace(namespaceName).execute(); } } + + @Test + void listWatchingNamespacesWithCachePredicate() throws Exception { + ApiClient client = ClientBuilder.defaultClient(); + CoreV1Api coreV1Api = new CoreV1Api(client); + SharedInformerFactory informerFactory = new SharedInformerFactory(client); + String selectedNamespace = "e2e-cache-selected"; + String ignoredNamespace = "e2e-cache-ignored"; + + coreV1Api + .createNamespace( + new V1Namespace() + .metadata( + new V1ObjectMeta() + .name(selectedNamespace) + .labels(java.util.Map.of("cache-filter", "keep")))) + .execute(); + coreV1Api + .createNamespace( + new V1Namespace() + .metadata( + new V1ObjectMeta() + .name(ignoredNamespace) + .labels(java.util.Map.of("cache-filter", "drop")))) + .execute(); + + GenericKubernetesApi api = + new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client); + SharedIndexInformer nsInformer = + informerFactory.sharedIndexInformerFor( + api, + V1Namespace.class, + 0, + ns -> + ns.getMetadata() != null + && ns.getMetadata().getLabels() != null + && "keep".equals(ns.getMetadata().getLabels().get("cache-filter"))); + + try { + informerFactory.startAllRegisteredInformers(); + await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue()); + await() + .untilAsserted( + () -> { + java.util.List cachedNamespaceNames = + nsInformer.getIndexer().list().stream() + .map(ns -> ns.getMetadata().getName()) + .collect(Collectors.toList()); + assertThat(cachedNamespaceNames).contains(selectedNamespace); + assertThat(cachedNamespaceNames).doesNotContain(ignoredNamespace); + }); + } finally { + informerFactory.stopAllRegisteredInformers(true); + coreV1Api.deleteNamespace(selectedNamespace).execute(); + coreV1Api.deleteNamespace(ignoredNamespace).execute(); + } + } } diff --git a/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java b/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java index f4fcac6e52..7c92ae3506 100644 --- a/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java +++ b/util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java @@ -37,6 +37,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.function.BiConsumer; +import java.util.function.Predicate; /** * SharedInformerFactory class constructs and caches informers for api types. @@ -200,6 +201,17 @@ SharedIndexInformer sharedIndexInformerFor( Class apiTypeClass, long resyncPeriodInMillis, BiConsumer, Throwable> exceptionHandler) { + return sharedIndexInformerFor( + listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, obj -> true); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + ListerWatcher listerWatcher, + Class apiTypeClass, + long resyncPeriodInMillis, + BiConsumer, Throwable> exceptionHandler, + Predicate cachePredicate) { Type apiType = TypeToken.get(apiTypeClass).getType(); if(informers.containsKey(apiType) && reuseExistingCachedInformer) { @@ -208,7 +220,11 @@ SharedIndexInformer sharedIndexInformerFor( SharedIndexInformer informer = new DefaultSharedIndexInformer<>( - apiTypeClass, listerWatcher, resyncPeriodInMillis, new Cache<>(), exceptionHandler); + apiTypeClass, + listerWatcher, + resyncPeriodInMillis, + new Cache<>(cachePredicate), + exceptionHandler); this.informers.putIfAbsent(apiType, informer); return informer; @@ -233,6 +249,21 @@ SharedIndexInformer sharedIndexInformerFor( genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, Namespaces.NAMESPACE_ALL); } + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + Predicate cachePredicate) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + Namespaces.NAMESPACE_ALL, + null, + cachePredicate); + } + /** * Working the same as {@link SharedInformerFactory#sharedIndexInformerFor} above. * @@ -253,7 +284,27 @@ SharedIndexInformer sharedIndexInformerFor( long resyncPeriodInMillis, String namespace) { return sharedIndexInformerFor( - genericKubernetesApi, apiTypeClass, resyncPeriodInMillis, namespace, null); + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + (BiConsumer, Throwable>) null); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + String namespace, + Predicate cachePredicate) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + null, + cachePredicate); } /** @@ -277,10 +328,27 @@ SharedIndexInformer sharedIndexInformerFor( long resyncPeriodInMillis, String namespace, BiConsumer, Throwable> exceptionHandler) { + return sharedIndexInformerFor( + genericKubernetesApi, + apiTypeClass, + resyncPeriodInMillis, + namespace, + exceptionHandler, + obj -> true); + } + + public synchronized + SharedIndexInformer sharedIndexInformerFor( + GenericKubernetesApi genericKubernetesApi, + Class apiTypeClass, + long resyncPeriodInMillis, + String namespace, + BiConsumer, Throwable> exceptionHandler, + Predicate cachePredicate) { ListerWatcher listerWatcher = listerWatcherFor(genericKubernetesApi, namespace); return sharedIndexInformerFor( - listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler); + listerWatcher, apiTypeClass, resyncPeriodInMillis, exceptionHandler, cachePredicate); } private diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java b/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java index 9b2f5ae96a..ba1f69b6c7 100644 --- a/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java +++ b/util/src/main/java/io/kubernetes/client/informer/cache/Cache.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.Set; import java.util.function.Function; +import java.util.function.Predicate; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.collections4.MapUtils; @@ -43,11 +44,22 @@ public class Cache implements Indexer /** indices stores objects' keys by their indices */ private Map>> indices = new HashMap<>(); + private Predicate storePredicate; + public Cache() { this( Caches.NAMESPACE_INDEX, Caches::metaNamespaceIndexFunc, - Caches::deletionHandlingMetaNamespaceKeyFunc); + Caches::deletionHandlingMetaNamespaceKeyFunc, + obj -> true); + } + + public Cache(Predicate storePredicate) { + this( + Caches.NAMESPACE_INDEX, + Caches::metaNamespaceIndexFunc, + Caches::deletionHandlingMetaNamespaceKeyFunc, + storePredicate); } /** @@ -61,9 +73,18 @@ public Cache( String indexName, Function> indexFunc, Function keyFunc) { + this(indexName, indexFunc, keyFunc, obj -> true); + } + + public Cache( + String indexName, + Function> indexFunc, + Function keyFunc, + Predicate storePredicate) { this.indexers.put(indexName, indexFunc); this.keyFunc = keyFunc; this.indices.put(indexName, new HashMap<>()); + this.storePredicate = storePredicate; } /** @@ -73,6 +94,9 @@ public Cache( */ @Override public void add(ApiType obj) { + if (!storePredicate.test(obj)) { + return; + } String key = keyFunc.apply(obj); synchronized (this) { ApiType oldObj = this.items.get(key); @@ -91,6 +115,13 @@ public void update(ApiType obj) { String key = keyFunc.apply(obj); synchronized (this) { ApiType oldObj = this.items.get(key); + if (!storePredicate.test(obj)) { + if (oldObj != null) { + this.deleteFromIndices(oldObj, key); + this.items.remove(key); + } + return; + } this.items.put(key, obj); updateIndices(oldObj, obj, key); } @@ -123,6 +154,9 @@ public void delete(ApiType obj) { public synchronized void replace(List list, String resourceVersion) { Map newItems = new HashMap<>(); for (ApiType item : list) { + if (!storePredicate.test(item)) { + continue; + } String key = keyFunc.apply(item); newItems.put(key, item); } diff --git a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java index 7c1690b66f..fe90e411e7 100644 --- a/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java +++ b/util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java @@ -301,11 +301,19 @@ public void handleDeltas(Deque nodeNameIndexedPods = podCache.byIndex(nodeIndex, "node1"); assertThat(nodeNameIndexedPods).hasSize(1); } + + @Test + void cachePredicateShouldFilterAddAndReplace() { + Cache podCache = + new Cache<>(pod -> "keep".equals(pod.getMetadata().getLabels().get("scope"))); + V1Pod kept = + new V1Pod() + .metadata( + new V1ObjectMeta() + .namespace("ns") + .name("kept") + .labels(Map.of("scope", "keep"))); + V1Pod dropped = + new V1Pod() + .metadata( + new V1ObjectMeta() + .namespace("ns") + .name("dropped") + .labels(Map.of("scope", "drop"))); + + podCache.add(kept); + podCache.add(dropped); + assertThat(podCache.list()).containsExactly(kept); + + podCache.replace(Arrays.asList(kept, dropped), "1"); + assertThat(podCache.list()).containsExactly(kept); + } + + @Test + void cachePredicateShouldEvictOnUpdateWhenObjectNoLongerMatches() { + Cache podCache = + new Cache<>(pod -> "keep".equals(pod.getMetadata().getLabels().get("scope"))); + V1Pod pod = + new V1Pod() + .metadata( + new V1ObjectMeta() + .namespace("ns") + .name("pod") + .labels(new HashMap<>(Map.of("scope", "keep")))); + + podCache.add(pod); + assertThat(podCache.list()).hasSize(1); + + pod.getMetadata().setLabels(Map.of("scope", "drop")); + podCache.update(pod); + + assertThat(podCache.list()).isEmpty(); + } } diff --git a/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java b/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java new file mode 100644 index 0000000000..041b73e29b --- /dev/null +++ b/util/src/test/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformerCachePredicateTest.java @@ -0,0 +1,64 @@ +/* +Copyright 2026 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.impl; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +import io.kubernetes.client.common.KubernetesObject; +import io.kubernetes.client.informer.ListerWatcher; +import io.kubernetes.client.informer.cache.Cache; +import io.kubernetes.client.informer.cache.DeltaFIFO; +import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.openapi.models.V1Pod; +import io.kubernetes.client.openapi.models.V1PodList; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.Map; +import org.apache.commons.lang3.tuple.MutablePair; +import org.junit.jupiter.api.Test; + +class DefaultSharedIndexInformerCachePredicateTest { + + private static V1Pod pod(String name, String scope) { + return new V1Pod() + .metadata( + new V1ObjectMeta().namespace("default").name(name).labels(Map.of("scope", scope))); + } + + @Test + void handleDeltasShouldRespectCachePredicateForAddAndUpdate() { + Cache cache = + new Cache<>(p -> "keep".equals(p.getMetadata().getLabels().get("scope"))); + DefaultSharedIndexInformer informer = + new DefaultSharedIndexInformer<>( + V1Pod.class, + mock(ListerWatcher.class), + 0, + cache); + + Deque> deltas = new ArrayDeque<>(); + deltas.add(MutablePair.of(DeltaFIFO.DeltaType.Added, pod("dropped", "drop"))); + deltas.add(MutablePair.of(DeltaFIFO.DeltaType.Added, pod("kept", "keep"))); + informer.handleDeltas(deltas); + + assertThat(cache.list()).extracting(p -> p.getMetadata().getName()).containsExactly("kept"); + + V1Pod becomesDropped = pod("kept", "drop"); + Deque> update = new ArrayDeque<>(); + update.add(MutablePair.of(DeltaFIFO.DeltaType.Updated, becomesDropped)); + informer.handleDeltas(update); + + assertThat(cache.list()).isEmpty(); + } +}