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

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

@@ -197,7 +197,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

@@ -108,11 +108,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;
   }
 
@@ -193,4 +203,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

@@ -130,7 +130,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,205 @@
+/*
+ * 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.
+ */
+@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 = refCount.decrementAndGet();
+      if (count < 0) {
+        throw new AssertionError("SharedResourceManager reference count dropped below 0");
+      }
+      if (count == 0) {
+        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 final Object closeLock = new Object();
+  private volatile boolean closed;
+
+  public SharedResourceManager(Function<K, V> resourceCreator) {
+    this.resourceCreator = resourceCreator;
+  }
+
+  /**
+   * Acquires a resource for the given key, incrementing its reference count.
+   */
+  public V acquire(K key) {
+    while (true) {
+      SharedResource<V> shared = resources.get(key);
+      if (shared == null) {
+        SharedResource<V> newShared = new SharedResource<>(resourceCreator.apply(key));
+        synchronized (closeLock) {
+          if (closed) {
+            newShared.release();
+            throw new IllegalStateException("SharedResourceManager is closed");
+          }
+          shared = resources.putIfAbsent(key, newShared);
+        }
+        if (shared == null) {
+          return newShared.get();
+        }
+        // Lost the race: close the resource we created to prevent leaks.
+        newShared.release();
+      }
+      if (shared.retain()) {
+        return shared.get();
+      }
+      // If retain failed, it's being evicted concurrently. Loop to compute a new one.
+      resources.remove(key, shared);
+    }
+  }
+
+  /**
+   * Releases a resource for the given key, decrementing its reference count.
+   * Closes and evicts the resource if the reference count reaches 0.
+   *
+   * <p>In the xDS filter state lifecycle, this method is invoked from cleanup
+   * tasks that execute within the {@code SynchronizationContext}.
+   *
+   * @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 reference for each.
+   *
+   * <p>This intentionally performs a single {@code release()} per entry, decrementing the
+   * manager's own reference count contribution. 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.
+   */
+  public void close() {
+    synchronized (closeLock) {
+      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
+        }
+      }
+    }
+  }
+}