Add allowWatchBookmarks and resourceVersionMatch support to GenericKubernetesApi list/watch

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,35 @@ 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<DynamicKubernetesListObject> listResponse =
+          dynamicApi.list(
+              ListOptions.builder()
+                  .fieldSelector("metadata.name=" + namespaceName)
+                  .resourceVersion(createdNamespace.getMetadata().getResourceVersion())
+                  .resourceVersionMatch("NotOlderThan")
+                  .allowWatchBookmarks(true)
+                  .build());
+
+      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();
+    }
+  }
 }

e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java

@@ -17,12 +17,25 @@
 
 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 java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
 
 class NamespaceInformerTest {
@@ -57,4 +70,181 @@ 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<V1Namespace, V1NamespaceList> api =
+        new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
+
+    SharedIndexInformer<V1Namespace> 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<V1Namespace>() {
+            @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.get()).isTrue();
+      assertThat(ignoredSeen.get()).isFalse();
+    } finally {
+      informerFactory.stopAllRegisteredInformers(true);
+      coreV1Api.deleteNamespace(selectedNamespace).execute();
+      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<V1Namespace, V1NamespaceList> api =
+        new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
+
+    ListerWatcher<V1Namespace, V1NamespaceList> flakyWatcher =
+        new ListerWatcher<V1Namespace, V1NamespaceList>() {
+          @Override
+          public V1NamespaceList list(CallGeneratorParams params) {
+            return api
+                .list(
+                    new ListOptions()
+                        .resourceVersion(params.resourceVersion)
+                        .timeoutSeconds(params.timeoutSeconds))
+                .getObject();
+          }
+
+          @Override
+          public Watchable<V1Namespace> 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<V1Namespace> nsInformer =
+        informerFactory.sharedIndexInformerFor(flakyWatcher, V1Namespace.class, 0);
+    CountDownLatch selectedSeen = new CountDownLatch(1);
+    try {
+      nsInformer.addEventHandler(
+          new ResourceEventHandler<V1Namespace>() {
+            @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();
+    }
+  }
+
+  @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<V1Namespace, V1NamespaceList> api =
+        new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
+    SharedIndexInformer<V1Namespace> 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<String> 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();
+    }
+  }
 }

util/src/main/java/io/kubernetes/client/informer/FilteringResourceEventHandler.java

@@ -0,0 +1,75 @@
+/*
+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<ApiType extends KubernetesObject>
+    implements ResourceEventHandler<ApiType> {
+
+  private final ResourceEventHandler<ApiType> delegate;
+  private final Predicate<ApiType> filter;
+
+  public FilteringResourceEventHandler(
+      ResourceEventHandler<ApiType> delegate, Predicate<ApiType> filter) {
+    this.delegate = Objects.requireNonNull(delegate);
+    this.filter = Objects.requireNonNull(filter);
+  }
+
+  @Override
+  public void onAdd(ApiType obj) {
+    if (matches(obj)) {
+      delegate.onAdd(obj);
+    }
+  }
+
+  @Override
+  public void onUpdate(ApiType oldObj, ApiType newObj) {
+    boolean oldMatched = matches(oldObj);
+    boolean newMatched = matches(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 (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;
+    }
+  }
+}

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<ApiType extends KubernetesObject> {
    */
   void addEventHandler(ResourceEventHandler<ApiType> handler);
 
+  /**
+   * Add event handler with predicate filter.
+   *
+   * @param handler the handler
+   * @param filter the object filter
+   */
+  void addEventHandler(ResourceEventHandler<ApiType> handler, Predicate<ApiType> 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<ApiType extends KubernetesObject> {
    */
   void addEventHandlerWithResyncPeriod(ResourceEventHandler<ApiType> 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<ApiType> handler, long resyncPeriod, Predicate<ApiType> filter);
+
   /** run starts the shared informer, which will be stopped until stop() is called. */
   void run();