[fix][admin] PIP-478: off-load the admin's v4 credential and lend it a bounded auth executor

pip/pip-478.md

@@ -1023,7 +1023,7 @@ The style is selected purely by which capability the plugin exposes — today on
 
 **Who owns the future a budget is applied to (normative).** The framework bounds each stage of the exchange by timing out a defensive **copy** of the stage future, never the future the plugin or the transport returned. `CompletableFuture.orTimeout` returns — and completes — *the receiver*, so bounding an SPI-supplied future in place would permanently fail a plugin that legitimately memoizes one shared credential future (the natural shape of an OAuth2 access-token cache): one slow request would turn into a client-wide authentication outage. A corollary follows from bounding the copy: the framework does **not** cancel the underlying work on timeout — deliberately, since cancelling a shared plugin future is the same hazard — so cleanup on timeout belongs to the stage itself, and both HTTP transports self-bound against the `Duration` they are handed (request timeout on the AsyncHttpClient one; the JAX-RS one times out its own future and cancels the in-flight request).
 
-**The budget covers the v4 branch too.** Both HTTP transports hand the driver an explicit budget — the lookup client's `lookupTimeoutMs` and the admin client's `requestTimeoutMs`, each falling back to 60 s — and the driver bounds each stage against the remaining budget, surfacing exhaustion as an authentication failure naming the round. On the HTTP-lookup client the v4 branch — taken when the plugin exposes no HTTP challenge capability — is not driven inline either: the whole v4 header composition is off-loaded to the blocking auth executor and composed with `thenCompose`, never joined, so a plugin whose `getAuthData()` refreshes a token cannot deadlock against the event loop that would serve that refresh. **The admin client's `BaseResource` is the exception**: it still composes v4 headers on the calling thread. That thread is the application's own, not an event loop, so it is a latency rather than a liveness problem — but it is the one place the "every synchronous v4 plugin call is off-loaded" rule does not yet hold, and it is tracked as a follow-up.
+**The budget covers the v4 branch too.** Both HTTP transports hand the driver an explicit budget — the lookup client's `lookupTimeoutMs` and the admin client's `requestTimeoutMs`, each falling back to 60 s — and the driver bounds each stage against the remaining budget, surfacing exhaustion as an authentication failure naming the round. On the HTTP-lookup client the v4 branch — taken when the plugin exposes no HTTP challenge capability — is not driven inline either: the whole v4 header composition is off-loaded to the blocking auth executor and composed with `thenCompose`, never joined, so a plugin whose `getAuthData()` refreshes a token cannot deadlock against the event loop that would serve that refresh. The admin client's `BaseResource` composes its v4 headers the same way, on a bounded authentication executor owned by the `PulsarAdmin` that lent it (falling back to the shared pool for a resource built outside one), so the "every synchronous v4 plugin call is off-loaded" rule holds on every transport.
 
 ### Class-name compatibility and the v4 internal migration
 

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java

@@ -39,6 +39,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -58,6 +59,7 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.auth.v5.AsyncHttpAuthenticationProvider;
 import org.apache.pulsar.client.impl.auth.v5.HttpAuthenticationDriver;
+import org.apache.pulsar.client.impl.auth.v5.V5AuthContexts;
 import org.apache.pulsar.common.policies.data.ErrorData;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 
@@ -69,12 +71,42 @@ public abstract class BaseResource {
 
     protected final Authentication auth;
     protected final long requestTimeoutMs;
+    // PIP-478: the owning admin's bounded blocking executor, where the deprecated v4 credential composition
+    // runs. Lent by PulsarAdminImpl right after construction rather than threaded through the constructors
+    // of all 23 resource classes, and never mutated afterwards. Null for a resource built outside a
+    // PulsarAdmin (tests, embedders), which has no admin pool to borrow and falls back to the framework's
+    // shared one — still off the caller thread.
+    private volatile Executor blockingAuthExecutor;
 
     protected BaseResource(Authentication auth, long requestTimeoutMs) {
         this.auth = auth;
         this.requestTimeoutMs = requestTimeoutMs;
     }
 
+    /**
+     * Lend this resource the owning admin's bounded blocking authentication executor (PIP-478), so a
+     * stalled identity provider reached through this admin's plugin cannot occupy the process-wide shared
+     * pool that every other client in the JVM depends on. Called once by {@link PulsarAdminImpl}, before
+     * the resource is published.
+     *
+     * @param blockingAuthExecutor the owning admin's blocking authentication executor
+     */
+    void setBlockingAuthExecutor(Executor blockingAuthExecutor) {
+        this.blockingAuthExecutor = blockingAuthExecutor;
+    }
+
+    /**
+     * The executor lent by {@link PulsarAdminImpl}, or {@code null} when none was (VisibleForTesting). The
+     * lending is opt-in per construction site, and a resource that misses it still works — it just falls
+     * back to the shared pool — so the only thing that can catch a resource added without it is a test that
+     * reads this.
+     *
+     * @return the lent blocking authentication executor, or {@code null}
+     */
+    Executor blockingAuthExecutorForTest() {
+        return blockingAuthExecutor;
+    }
+
     public Builder request(final WebTarget target) throws PulsarAdminException {
         try {
             return requestAsync(target).get();
@@ -123,7 +155,7 @@ public CompletableFuture<Builder> requestAsync(final WebTarget target) {
      * {@code GET} to the original URI each round, exactly what the v4 {@code authenticationStage(...)} does
      * today — and yields the validated role-token headers. Otherwise the deprecated v4
      * {@code authenticationStage(...)} / {@code newRequestHeader(...)} hooks run verbatim, preserving
-     * behaviour for third-party plugins and single-pass built-ins.
+     * behaviour for third-party plugins and single-pass built-ins — but off the calling thread.
      */
     protected CompletableFuture<Map<String, String>> computeAuthHeaders(URI uri) {
         try {
@@ -137,13 +169,50 @@ protected CompletableFuture<Map<String, String>> computeAuthHeaders(URI uri) {
                             .thenApply(headers -> (headers == null || headers.isEmpty()) ? null : headers.asMap());
                 }
             }
+            // The deprecated v4 hooks may block — an OAuth2 or Athenz shim's getAuthData() refreshes its
+            // credential with a synchronous HTTP exchange — and this method runs on whatever thread issued the
+            // admin call, which for a broker calling its own admin client is a request-handling thread.
+            // HttpClient.computeAuthHeaders off-loads the identical composition for the lookup path and names
+            // the self-deadlock it avoids; this was the last caller-thread credential resolution left, and the
+            // one place PIP-478's "every synchronous v4 plugin call is off-loaded" was still an overstatement.
+            //
+            // It runs on the owning admin's own bounded pool, the same one a services-aware plugin is lent, so
+            // the work that can actually stall — a KDC or IdP round trip inside the plugin — stays isolated
+            // per admin: were this the framework's process-wide shared pool, one admin's stalled provider
+            // would throttle authentication for every other client in the JVM. A resource built outside a
+            // PulsarAdmin has no pool to borrow and falls back to that shared one, which is still off the
+            // caller thread.
+            Executor executor = V5AuthContexts.blockingExecutorOrShared(blockingAuthExecutor);
+            return V5AuthContexts.supplyBlocking(executor, () -> v4AuthHeaders(uri, executor))
+                    .thenCompose(headers -> headers);
+        } catch (Throwable t) {
+            return CompletableFuture.failedFuture(t);
+        }
+    }
+
+    /**
+     * The deprecated v4 HTTP authentication composition, run on a blocking executor by
+     * {@link #computeAuthHeaders(URI)}.
+     *
+     * @param uri the request URI
+     * @param blockingExecutor the executor this composition runs on, for the continuation below
+     * @return a future of the headers, or of {@code null} when the plugin contributes none
+     */
+    private CompletableFuture<Map<String, String>> v4AuthHeaders(URI uri, Executor blockingExecutor) {
+        try {
             AuthenticationDataProvider authData = auth.getAuthData(uri.getHost());
             if (!authData.hasDataForHttp()) {
                 return CompletableFuture.completedFuture(null);
             }
             CompletableFuture<Map<String, String>> stage = new CompletableFuture<>();
             auth.authenticationStage(uri.toString(), authData, null, stage);
-            return stage.thenApply(respHeaders -> {
+            // thenApplyAsync, not thenApply: newRequestHeader is the second synchronous v4 hook, and a plugin
+            // that completes the stage asynchronously — the multi-round challenge shape — completes it from
+            // its own HTTP callback thread. A plain continuation would run that hook there, which off-loading
+            // the resolution alone would not cover. In-tree that path now belongs to the v5 driver above, so
+            // this is for third-party v4 plugins; where the stage completes inline (the single-pass default)
+            // the hop is to a sibling task on this same executor.
+            return stage.thenApplyAsync(respHeaders -> {
                 try {
                     Set<Entry<String, String>> headers = auth.newRequestHeader(uri.toString(), authData, respHeaders);
                     if (headers == null) {
@@ -155,7 +224,7 @@ protected CompletableFuture<Map<String, String>> computeAuthHeaders(URI uri) {
                 } catch (Exception e) {
                     throw new CompletionException(e);
                 }
-            });
+            }, blockingExecutor);
         } catch (Throwable t) {
             return CompletableFuture.failedFuture(t);
         }