feat(xds): Add filter state reference-counted resource management

- Introduce `SharedResourceManager` to manage reference-counted shared
  resources across xDS filters and resolvers.
- Update `XdsNameResolver` to integrate with `SharedResourceManager` and
  support filter state retention.
- Add `RefCountedRoute` and `RefCountedRouteInterceptor` to ensure
  resources are reliably retained during call interception and released
  across all RPC lifecycle termination paths (normal completion,
  cancellation, and exceptions).
- Update `Filter` interfaces and existing implementations
  (`FaultFilter`, `GcpAuthenticationFilter`) to accommodate filter state
  sharing.

Author: sauravzg

xds/src/main/java/io/grpc/xds/ExternalProcessorFilter.java

@@ -128,7 +128,8 @@ public ConfigOrError<ExternalProcessorFilterOverrideConfig> parseFilterConfigOve
   @Nullable
   @Override
   public ClientInterceptor buildClientInterceptor(FilterConfig filterConfig,
-      @Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler) {
+      @Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler,
+      Filter.ResourceCleanupRegistry cleanupRegistry) {
     ExternalProcessorFilterConfig extProcFilterConfig =
         (ExternalProcessorFilterConfig) filterConfig;
     if (overrideConfig != null) {

xds/src/main/java/io/grpc/xds/FaultFilter.java

@@ -199,7 +199,8 @@ private static FaultConfig.FractionalPercent parsePercent(FractionalPercent prot
   @Override
   public ClientInterceptor buildClientInterceptor(
       FilterConfig config, @Nullable FilterConfig overrideConfig,
-      final ScheduledExecutorService scheduler) {
+      final ScheduledExecutorService scheduler,
+      Filter.ResourceCleanupRegistry cleanupRegistry) {
     checkNotNull(config, "config");
     if (overrideConfig != null) {
       config = overrideConfig;

xds/src/main/java/io/grpc/xds/Filter.java

@@ -109,11 +109,21 @@ ConfigOrError<? extends FilterConfig> parseFilterConfigOverride(
         Message rawProtoMessage, FilterConfigParseContext context);
   }
 
-  /** Uses the FilterConfigs produced above to produce an HTTP filter interceptor for clients. */
+  /**
+   * Builds an HTTP filter interceptor for this route.
+   *
+   * <p>Filters that create stateful resources (e.g., shared channels) should register
+   * cleanup tasks via {@code cleanupRegistry}. These tasks execute in the xDS
+   * {@code SynchronizationContext} when the route's reference count reaches zero,
+   * meaning no in-flight RPCs reference the route and the control plane has released it.
+   *
+   * @param cleanupRegistry registry for cleanup tasks; never null
+   */
   @Nullable
   default ClientInterceptor buildClientInterceptor(
       FilterConfig config, @Nullable FilterConfig overrideConfig,
-      ScheduledExecutorService scheduler) {
+      ScheduledExecutorService scheduler,
+      ResourceCleanupRegistry cleanupRegistry) {
     return null;
   }
 
@@ -206,4 +216,15 @@ public String toString() {
           .toString();
     }
   }
+
+  /**
+   * Registry for cleanup tasks associated with a route's resource scope.
+   */
+  @FunctionalInterface
+  interface ResourceCleanupRegistry {
+    /**
+     * Registers a task to run when the route is no longer in use.
+     */
+    void addCleanupTask(Runnable task);
+  }
 }

xds/src/main/java/io/grpc/xds/GcpAuthenticationFilter.java

@@ -132,7 +132,8 @@ public ConfigOrError<GcpAuthenticationConfig> parseFilterConfigOverride(
   @Nullable
   @Override
   public ClientInterceptor buildClientInterceptor(FilterConfig config,
-      @Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler) {
+      @Nullable FilterConfig overrideConfig, ScheduledExecutorService scheduler,
+      Filter.ResourceCleanupRegistry cleanupRegistry) {
 
     ComputeEngineCredentials credentials = ComputeEngineCredentials.create();
     synchronized (callCredentialsCache) {

xds/src/main/java/io/grpc/xds/SharedResourceManager.java

@@ -0,0 +1,207 @@
+/*
+ * Copyright 2026 The gRPC 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.grpc.xds;
+
+import com.google.common.base.Preconditions;
+import io.grpc.Internal;
+import io.grpc.ManagedChannel;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import javax.annotation.concurrent.ThreadSafe;
+
+/**
+ * Manages generic reference-counted shared resources for xDS filters.
+ *
+ * <p>Similar to {@code io.grpc.xds.internal.security.ReferenceCountingMap}, but provides
+ * additional lifecycle management ({@link #close()}) and a simpler key-only
+ * {@link #release(Object)} API designed for xDS filter state cleanup tasks.
+ *
+ * <h3>Requirements</h3>
+ * <ul>
+ *   <li>{@link #acquire} and {@link #close} must be called from the same serialized context
+ *       (e.g., {@code SynchronizationContext}). They must never race each other.</li>
+ *   <li>{@link #release} may be called from any thread.</li>
+ *   <li>Managed resources ({@link ResourceCloseable#close}) must be thread-safe, as they may
+ *       be invoked from any thread when the last reference is released.</li>
+ *   <li>Keys are canonical resource identifiers: a given key always maps to the same logical
+ *       resource. The manager will never hold two distinct resources for the same key.</li>
+ * </ul>
+ */
+@Internal
+public final class SharedResourceManager<K, V extends SharedResourceManager.ResourceCloseable> {
+
+  /**
+   * An AutoCloseable resource that explicitly guarantees its close operation
+   * will not throw checked exceptions.
+   */
+  public interface ResourceCloseable extends AutoCloseable {
+    @Override
+    void close();
+  }
+
+  /**
+   * Adapts {@link ManagedChannel} to {@link ResourceCloseable} for management by
+   * {@link SharedResourceManager}.
+   */
+  public static final class ManagedChannelResource implements ResourceCloseable {
+    private final ManagedChannel channel;
+
+    public ManagedChannelResource(ManagedChannel channel) {
+      this.channel = Preconditions.checkNotNull(channel, "channel");
+    }
+
+    @Override
+    public void close() {
+      channel.shutdown();
+    }
+
+    public ManagedChannel getChannel() {
+      return channel;
+    }
+  }
+
+  /**
+   * An internal pure reference-counting container managing a stateful ResourceCloseable.
+   */
+  @ThreadSafe
+  static final class SharedResource<T extends ResourceCloseable> {
+    private final T resource;
+    private final AtomicInteger refCount = new AtomicInteger(1);
+
+    SharedResource(T resource) {
+      this.resource = Preconditions.checkNotNull(resource, "resource");
+    }
+
+    /**
+     * Retains the resource. Returns false if the resource has hit 0 and is being closed.
+     */
+    boolean retain() {
+      int count;
+      do {
+        count = refCount.get();
+        if (count == 0) {
+          return false;
+        }
+      } while (!refCount.compareAndSet(count, count + 1));
+      return true;
+    }
+
+    /**
+     * Decrements reference count. Closes underlying resource if count hits 0.
+     * @return true if the count reached 0 and the resource was closed; false otherwise.
+     */
+    boolean release() {
+      int count;
+      do {
+        count = refCount.get();
+        if (count <= 0) {
+          throw new AssertionError("SharedResourceManager reference count is already 0");
+        }
+      } while (!refCount.compareAndSet(count, count - 1));
+      if (count == 1) {
+        resource.close();
+        return true;
+      }
+      return false;
+    }
+
+    T get() {
+      return resource;
+    }
+
+    int getRefCount() {
+      return refCount.get();
+    }
+  }
+
+  private final ConcurrentMap<K, SharedResource<V>> resources = new ConcurrentHashMap<>();
+  private final Function<K, V> resourceCreator;
+  private boolean closed;
+
+  public SharedResourceManager(Function<K, V> resourceCreator) {
+    this.resourceCreator = resourceCreator;
+  }
+
+  /**
+   * Acquires a resource for the given key, incrementing its reference count.
+   *
+   * <p>Must be called from the {@code SynchronizationContext}.
+   */
+  public V acquire(K key) {
+    Preconditions.checkState(!closed, "SharedResourceManager is closed");
+    SharedResource<V> shared = resources.computeIfAbsent(key,
+        k -> new SharedResource<>(resourceCreator.apply(k)));
+    Preconditions.checkState(shared.retain(),
+        "retain() failed; close() should not have run");
+    return shared.get();
+  }
+
+  /**
+   * Releases a resource for the given key, decrementing its reference count.
+   * Closes and evicts the resource if the reference count reaches 0.
+   *
+   * <p>Thread-safe: may be called from any thread. The underlying {@link SharedResource}
+   * uses CAS-based reference counting, and eviction uses
+   * {@link ConcurrentMap#remove(Object, Object)} for safe concurrent removal.
+   *
+   * <p>This API takes only a key (not the resource value) because keys are canonical
+   * resource identifiers (see class-level requirements).
+   *
+   * @return true if the resource was closed; false otherwise.
+   */
+  public boolean release(K key) {
+    SharedResource<V> shared = resources.get(key);
+    if (shared == null) {
+      return false;
+    }
+    try {
+      if (shared.release()) {
+        return resources.remove(key, shared);
+      }
+    } catch (Throwable t) {
+      resources.remove(key, shared);
+      throw t;
+    }
+    return false;
+  }
+
+  /**
+   * Removes all entries from the cache and releases the manager's creation reference for each.
+   *
+   * <p>This performs a single {@code release()} per entry, decrementing the manager's own
+   * reference count contribution (the initial refCount=1 from creation). If in-flight RPCs
+   * still hold references, the underlying resource remains open until those references are
+   * released. This avoids pulling resources out from under active operations.
+   *
+   * <p>Must be called from the {@code SynchronizationContext}.
+   */
+  public void close() {
+    closed = true;
+    for (K key : resources.keySet()) {
+      SharedResource<V> shared = resources.remove(key);
+      if (shared != null) {
+        try {
+          shared.release();
+        } catch (Throwable t) {
+          // Ignore exceptions during final close-all to ensure we try to close other resources
+        }
+      }
+    }
+  }
+}