From 023f2786d3cac3db6e979bc41b5a060ec53490e1 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 13 Aug 2026 20:03:10 +0300 Subject: [PATCH 1/4] [fix][admin] PIP-478: off-load the admin's v4 credential and lend it a bounded auth executor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PIP-478 states that every synchronous v4 plugin call is off-loaded, and after the core migration that was true everywhere except one place: BaseResource resolved the credential for each admin request on whatever thread issued the call. HttpClient.computeAuthHeaders off-loads the identical composition for the lookup path and names the self-deadlock it avoids — an OAuth2 or Athenz shim's getAuthData() refreshes its credential over synchronous HTTP, and for a broker calling its own admin client the caller is a request-handling thread. The composition now runs on a blocking executor, and produces the plugin's headers verbatim. The v4 composition is off-loaded to the framework's shared pool rather than the admin's own, because BaseResource is constructed by 30 subclasses that would each have to thread an executor through; the shared pool is exactly what PIP-478 provides for a component with none to lend. What a services-aware plugin gets is different, and is the second half of this change. PulsarAdminImpl bound its framework services with a null blocking executor. That stopped being a live defect when the init context grew a shared-pool fallback, but it left the SASL-over-HTTP challenge rounds putting their GSSAPI work on one process-wide pool shared with every other client in the JVM, so a stalled KDC reached by one admin throttled authentication for all of them. The admin now lends a small bounded pool of its own, created lazily and shut down with the admin, which keeps that blast radius inside the admin that owns the plugin. Its request threads are still never lent out — that was the hazard the original null was avoiding. The scheduler stays unbound: nothing on the admin path schedules periodic authentication work. Instead the bound init context now falls back to the shared scheduler and blocking pool when the bound services leave one null, so the SPI's "never null" contract holds on every path — previously, binding partial services was worse for a third-party plugin than binding none at all, since the no-services context has always guaranteed non-null. --- .../client/admin/internal/BaseResource.java | 31 +++- .../admin/internal/PulsarAdminImpl.java | 60 +++++++- .../admin/internal/AdminAuthOffloadTest.java | 134 ++++++++++++++++++ .../client/impl/auth/v5/V5AuthContexts.java | 11 +- .../auth/v5/BoundInitContextFallbackTest.java | 68 +++++++++ 5 files changed, 295 insertions(+), 9 deletions(-) create mode 100644 pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java create mode 100644 pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BoundInitContextFallbackTest.java diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java index feee6417c21b7..950b6814f3577 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java @@ -58,6 +58,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; @@ -123,7 +124,7 @@ public CompletableFuture 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> computeAuthHeaders(URI uri) { try { @@ -137,6 +138,34 @@ protected CompletableFuture> 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. + // + // A null executor sends the work to the framework's shared blocking pool rather than running it + // inline. That is deliberate: BaseResource is constructed by 30 subclasses that would each have to + // thread an executor through, and the shared pool is exactly what PIP-478 provides for a component + // with none to lend. The admin's own bounded pool is still what a services-aware plugin gets, so + // the work that can actually stall — a KDC or IdP round trip inside the plugin — stays isolated + // per admin. + return V5AuthContexts.supplyBlocking(null, () -> v4AuthHeaders(uri)).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 + * @return a future of the headers, or of {@code null} when the plugin contributes none + */ + private CompletableFuture> v4AuthHeaders(URI uri) { + try { AuthenticationDataProvider authData = auth.getAuthData(uri.getHost()); if (!authData.hasDataForHttp()) { return CompletableFuture.completedFuture(null); diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java index 473ab781e9da8..b607de03964d3 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java @@ -29,6 +29,9 @@ import java.time.Clock; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.Executor; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import lombok.CustomLog; import lombok.Getter; @@ -90,6 +93,10 @@ public class PulsarAdminImpl implements PulsarAdmin { public static final int DEFAULT_REQUEST_TIMEOUT_SECONDS = 300; + // PIP-478: ceiling for the admin's blocking authentication pool. Smaller than the client's 16 — + // an admin issues REST calls, not a fan-out of connection attempts against many brokers. + private static final int AUTH_BLOCKING_MAX_THREADS = 8; + private final Clusters clusters; private final Brokers brokers; private final BrokerStats brokerStats; @@ -124,6 +131,10 @@ public class PulsarAdminImpl implements PulsarAdmin { // (e.g. OAuth2) so an admin-only client acquires tokens over the framework client; null when the auth // plugin does not implement ClientAuthenticationServicesAware. Closed with this admin. private FrameworkHttpClientFactory authHttpClientFactory; + // PIP-478: the admin's own bounded executor for potentially-blocking authentication work — the v4 + // credential composition on every request, and whatever a services-aware plugin off-loads. Created + // lazily so an admin whose plugin never needs one pays nothing, and shut down with this admin. + private volatile ThreadPoolExecutor blockingAuthExecutor; @Getter private AsyncHttpConnectorProvider asyncConnectorProvider; @@ -595,16 +606,46 @@ private void bindAuthenticationServices(ClientConfigurationData conf) { () -> null, () -> null, () -> null, () -> authTlsFactory(conf), conf, clientInstanceId); OpenTelemetry openTelemetry = conf.getOpenTelemetry() != null ? conf.getOpenTelemetry() : OpenTelemetry.noop(); - // No scheduler: nothing on the admin path schedules periodic authentication work. The blocking - // executor is left unbound deliberately too — a plugin that needs one falls back to the shared pool - // rather than running on the caller thread, which is where the SASL-over-HTTP challenge rounds - // off-load their GSSAPI work. Handing the admin's own request threads over instead would let a slow - // KDC consume them. + // No scheduler: nothing on the admin path schedules periodic authentication work, and a plugin that + // does schedule some gets the framework's shared one — binding a scheduled pool per admin to sit idle + // would cost more than it buys. + // + // The blocking executor *is* the admin's own. This is where the SASL-over-HTTP challenge rounds put + // their GSSAPI work, so it must never be the admin's request threads — a slow KDC would consume them + // — but leaving it unbound is not right either: the plugin then shares one process-wide pool with + // every other client in the JVM, so a stalled identity provider reached by one admin throttles + // authentication for all of them. A small bounded pool per admin keeps that blast radius inside the + // admin that owns the plugin, and is shut down with it. ClientAuthenticationServices services = new DefaultClientAuthenticationServices( - authHttpClientFactory, null, null, Clock.systemDefaultZone(), openTelemetry, clientInstanceId); + authHttpClientFactory, null, blockingAuthExecutor(), Clock.systemDefaultZone(), openTelemetry, + clientInstanceId); aware.bindClientAuthenticationServices(services); } + /** + * The admin's bounded executor for potentially-blocking authentication work (PIP-478). + * + *

Queues rather than rejects: every caller is an authenticated request, so a saturated pool must slow + * requests down rather than fail them. Core threads time out, so an admin whose plugin never blocks pays + * for nothing. Created lazily under this admin's monitor and shut down in {@link #close()}. + * + * @return the blocking authentication executor + */ + private synchronized Executor blockingAuthExecutor() { + if (blockingAuthExecutor == null) { + ThreadPoolExecutor executor = new ThreadPoolExecutor(AUTH_BLOCKING_MAX_THREADS, + AUTH_BLOCKING_MAX_THREADS, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), + runnable -> { + Thread thread = new Thread(runnable, "pulsar-admin-auth-blocking"); + thread.setDaemon(true); + return thread; + }); + executor.allowCoreThreadTimeOut(true); + blockingAuthExecutor = executor; + } + return blockingAuthExecutor; + } + @Override public void close() { try { @@ -623,6 +664,13 @@ public void close() { // factory first would tear it down while those subscriptions are still live, which a custom factory // is entitled to treat as an error. asyncConnectorProvider.close(); + // PIP-478: after auth.close(), so a plugin shutting down over this executor still has it. shutdown() + // rather than shutdownNow(): queued work is a credential call for a request already in flight, and + // the threads are daemons, so a straggler cannot hold the JVM up. + ThreadPoolExecutor authExecutor = blockingAuthExecutor; + if (authExecutor != null) { + authExecutor.shutdown(); + } } @VisibleForTesting diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java new file mode 100644 index 0000000000000..9632c9cb8fcc8 --- /dev/null +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.pulsar.client.admin.internal; + +import static org.assertj.core.api.Assertions.assertThat; +import java.net.URI; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.pulsar.client.api.Authentication; +import org.apache.pulsar.client.api.AuthenticationDataProvider; +import org.apache.pulsar.client.api.PulsarClientException; +import org.testng.annotations.Test; + +/** + * PIP-478: the admin's v4 authentication composition must not run on the thread that issued the admin + * call. + * + *

{@code HttpClient} (the lookup path) has off-loaded the identical composition since the core + * migration, naming the self-deadlock it avoids; {@code BaseResource} was the one place left where a + * blocking {@code getAuthData()} — an OAuth2 or Athenz shim refreshing its credential over synchronous + * HTTP — ran wherever the caller happened to be, which for a broker calling its own admin client is a + * request-handling thread. + */ +public class AdminAuthOffloadTest { + + /** + * A v4 plugin whose credential call records the thread it ran on, standing in for one that blocks + * there. + */ + private static class ThreadRecordingAuthentication implements Authentication { + + private final AtomicReference getAuthDataThread = new AtomicReference<>(); + + @Override + public String getAuthMethodName() { + return "thread-recording"; + } + + @Override + public AuthenticationDataProvider getAuthData() { + getAuthDataThread.set(Thread.currentThread()); + return new AuthenticationDataProvider() { + @Override + public boolean hasDataForHttp() { + return true; + } + + @Override + public Set> getHttpHeaders() { + return Set.of(Map.entry("X-Test", "value")); + } + }; + } + + @Override + public void configure(Map authParams) { + } + + @Override + public void start() throws PulsarClientException { + } + + @Override + public void close() { + } + } + + /** The minimal concrete resource needed to reach the protected header composition. */ + private static class TestResource extends BaseResource { + TestResource(Authentication auth) { + super(auth, 30_000L); + } + + CompletableFuture> headers(URI uri) { + return computeAuthHeaders(uri); + } + } + + @Test + public void v4CredentialResolutionRunsOffTheCallerThread() throws Exception { + ThreadRecordingAuthentication auth = new ThreadRecordingAuthentication(); + TestResource resource = new TestResource(auth); + + Map headers = resource.headers(URI.create("http://broker.example:8080/admin/v2/clusters")) + .get(30, TimeUnit.SECONDS); + + // The composition still produces the plugin's headers verbatim — off-loading must not change what + // a third-party v4 plugin contributes. + assertThat(headers).containsExactly(Map.entry("X-Test", "value")); + assertThat(auth.getAuthDataThread.get()) + .as("the v4 credential must be resolved off the thread that issued the admin call") + .isNotNull() + .isNotSameAs(Thread.currentThread()); + } + + @Test + public void aPluginContributingNoHttpDataStillYieldsNoHeaders() throws Exception { + Authentication auth = new ThreadRecordingAuthentication() { + @Override + public AuthenticationDataProvider getAuthData() { + return new AuthenticationDataProvider() { + @Override + public boolean hasDataForHttp() { + return false; + } + }; + } + }; + + Map headers = new TestResource(auth) + .headers(URI.create("http://broker.example:8080/admin/v2/clusters")).get(30, TimeUnit.SECONDS); + + assertThat(headers).isNull(); + } +} diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java index 23baef8960485..96a32278d8080 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java @@ -242,12 +242,19 @@ public PulsarHttpClientFactory httpClientFactory() { @Override public ScheduledExecutorService scheduler() { - return services.scheduler(); + // Same fallback as the unbound context, and for the same reason: a component may bind services + // while leaving an accessor it has no use for null — the admin binds no scheduler, because + // nothing on its own path schedules periodic authentication work. Without this, binding + // *partial* services would be worse for a plugin than binding none at all, since the unbound + // context guarantees non-null. The SPI's "never null" contract now holds on every path. + ScheduledExecutorService scheduler = services.scheduler(); + return scheduler == null ? SharedScheduler.INSTANCE : scheduler; } @Override public Executor blockingExecutor() { - return services.blockingExecutor(); + Executor executor = services.blockingExecutor(); + return executor == null ? sharedBlockingExecutor() : executor; } @Override diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BoundInitContextFallbackTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BoundInitContextFallbackTest.java new file mode 100644 index 0000000000000..306f5f8d91063 --- /dev/null +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BoundInitContextFallbackTest.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.pulsar.client.impl.auth.v5; + +import static org.assertj.core.api.Assertions.assertThat; +import io.opentelemetry.api.OpenTelemetry; +import java.time.Clock; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext; +import org.testng.annotations.Test; + +/** + * PIP-478: {@link AuthenticationInitContext#scheduler()} and + * {@link AuthenticationInitContext#blockingExecutor()} are documented as never {@code null}, and the + * no-services context honours that. A component may still bind partial services — the admin binds + * an HTTP client factory and a blocking executor but no scheduler, because nothing on its path schedules + * periodic authentication work — and without a fallback on the bound path, binding partial services would + * be worse for a plugin than binding none at all. + */ +public class BoundInitContextFallbackTest { + + @Test + public void aBoundContextWithNoSchedulerStillSuppliesOne() { + AuthenticationInitContext ctx = V5AuthContexts.initContext( + new DefaultClientAuthenticationServices(null, null, null, Clock.systemUTC(), + OpenTelemetry.noop(), "test-client"), + "unused"); + + assertThat(ctx.scheduler()).as("a plugin scheduling a credential refresh must not NPE").isNotNull(); + assertThat(ctx.blockingExecutor()).as("the SPI tells plugins to off-load here").isNotNull(); + } + + @Test + public void boundExecutorsAreUsedWhenSupplied() { + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + Executor blocking = Runnable::run; + try { + AuthenticationInitContext ctx = V5AuthContexts.initContext( + new DefaultClientAuthenticationServices(null, scheduler, blocking, Clock.systemUTC(), + OpenTelemetry.noop(), "test-client"), + "unused"); + + // The fallback must not shadow what a component did bind. + assertThat(ctx.scheduler()).isSameAs(scheduler); + assertThat(ctx.blockingExecutor()).isSameAs(blocking); + } finally { + scheduler.shutdownNow(); + } + } +} From bf9941f8a02675ab883bf0db8bf8db0e6b1ee2ca Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Fri, 14 Aug 2026 18:23:15 +0300 Subject: [PATCH 2/4] [fix][admin] PIP-478: review fixes for the admin authentication executor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #26327 by Copilot. The admin built a bounded pool for blast-radius isolation and then sent its own per-request v4 credential composition to the framework's process-wide shared pool, so the common path — every authenticated admin request — kept the blast radius the pool exists to contain. The stated reason was cost: BaseResource is constructed 23 times and each would have to thread an executor through. Lending the pool right after construction gets the same result without touching the 23 resource classes. Second hook, same gap: newRequestHeader ran on whatever thread completed the authentication stage. For a v4 plugin of the multi-round challenge shape that is its own HTTP callback thread, so off-loading the credential resolution alone covered half the composition. The continuation now runs on the same blocking executor. HttpClient's lookup-path composition, which this one mirrors, had the identical gap and is fixed with it. The pool is now created with the admin rather than lazily: it is on the default path, an unused ThreadPoolExecutor starts no threads, and a final field removes the synchronized accessor. Tests: theV4CompositionRunsOnTheOwningAdminsOwnPool builds a real PulsarAdmin, so what is pinned is that the admin lends its pool — not that BaseResource would use one if given it. theNewRequestHeaderHookIsOffLoadedToo hands the stage back to the test to complete, because a stage completed before the continuation is registered runs it on the registering thread and would pass either way. Both mutation-verified: each fails alone when its fix is reverted. Assisted-by: Claude Code (Opus 5) --- .../client/admin/internal/BaseResource.java | 46 +++++-- .../admin/internal/PulsarAdminImpl.java | 125 ++++++++++-------- .../admin/internal/AdminAuthOffloadTest.java | 95 +++++++++++++ .../apache/pulsar/client/impl/HttpClient.java | 9 +- .../client/impl/auth/v5/V5AuthContexts.java | 17 ++- 5 files changed, 225 insertions(+), 67 deletions(-) diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java index 950b6814f3577..9296dcef07ea2 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java +++ b/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; @@ -70,12 +71,30 @@ 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; + } + public Builder request(final WebTarget target) throws PulsarAdminException { try { return requestAsync(target).get(); @@ -145,13 +164,15 @@ protected CompletableFuture> computeAuthHeaders(URI uri) { // 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. // - // A null executor sends the work to the framework's shared blocking pool rather than running it - // inline. That is deliberate: BaseResource is constructed by 30 subclasses that would each have to - // thread an executor through, and the shared pool is exactly what PIP-478 provides for a component - // with none to lend. The admin's own bounded pool is still what a services-aware plugin gets, so + // 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. - return V5AuthContexts.supplyBlocking(null, () -> v4AuthHeaders(uri)).thenCompose(headers -> headers); + // 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); } @@ -162,9 +183,10 @@ protected CompletableFuture> computeAuthHeaders(URI uri) { * {@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> v4AuthHeaders(URI uri) { + private CompletableFuture> v4AuthHeaders(URI uri, Executor blockingExecutor) { try { AuthenticationDataProvider authData = auth.getAuthData(uri.getHost()); if (!authData.hasDataForHttp()) { @@ -172,7 +194,13 @@ private CompletableFuture> v4AuthHeaders(URI uri) { } CompletableFuture> 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> headers = auth.newRequestHeader(uri.toString(), authData, respHeaders); if (headers == null) { @@ -184,7 +212,7 @@ private CompletableFuture> v4AuthHeaders(URI uri) { } catch (Exception e) { throw new CompletionException(e); } - }); + }, blockingExecutor); } catch (Throwable t) { return CompletableFuture.failedFuture(t); } diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java index b607de03964d3..3f195210d5b53 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java @@ -29,7 +29,6 @@ import java.time.Clock; import java.util.LinkedHashMap; import java.util.Map; -import java.util.concurrent.Executor; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -132,9 +131,11 @@ public class PulsarAdminImpl implements PulsarAdmin { // plugin does not implement ClientAuthenticationServicesAware. Closed with this admin. private FrameworkHttpClientFactory authHttpClientFactory; // PIP-478: the admin's own bounded executor for potentially-blocking authentication work — the v4 - // credential composition on every request, and whatever a services-aware plugin off-loads. Created - // lazily so an admin whose plugin never needs one pays nothing, and shut down with this admin. - private volatile ThreadPoolExecutor blockingAuthExecutor; + // credential composition on every request, and whatever a services-aware plugin off-loads. Shut down + // with this admin. Constructing it costs one object: a ThreadPoolExecutor starts no threads until a + // task arrives, and with core threads timing out an admin that issues no authenticated request keeps + // none. + private final ThreadPoolExecutor blockingAuthExecutor; @Getter private AsyncHttpConnectorProvider asyncConnectorProvider; @@ -166,6 +167,8 @@ public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigDa asyncConnectorProvider = new AsyncHttpConnectorProvider(clientConfigData, clientConfigData.getAutoCertRefreshSeconds(), acceptGzipCompression); + this.blockingAuthExecutor = newBlockingAuthExecutor(); + boolean constructed = false; try { bindAuthenticationServices(clientConfigData); @@ -204,29 +207,29 @@ public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigDa clientConfigData.getAutoCertRefreshSeconds(), sharedResources); long requestTimeoutMs = clientConfigData.getRequestTimeoutMs(); - this.clusters = new ClustersImpl(root, auth, requestTimeoutMs); - this.brokers = new BrokersImpl(root, auth, requestTimeoutMs); - this.brokerStats = new BrokerStatsImpl(root, auth, requestTimeoutMs); - this.proxyStats = new ProxyStatsImpl(root, auth, requestTimeoutMs); - this.tenants = new TenantsImpl(root, auth, requestTimeoutMs); - this.resourcegroups = new ResourceGroupsImpl(root, auth, requestTimeoutMs); - this.namespaces = new NamespacesImpl(root, auth, requestTimeoutMs); - this.topics = new TopicsImpl(root, auth, requestTimeoutMs); - this.localTopicPolicies = new TopicPoliciesImpl(root, auth, requestTimeoutMs, false); - this.globalTopicPolicies = new TopicPoliciesImpl(root, auth, requestTimeoutMs, true); - this.nonPersistentTopics = new NonPersistentTopicsImpl(root, auth, requestTimeoutMs); - this.resourceQuotas = new ResourceQuotasImpl(root, auth, requestTimeoutMs); - this.lookups = new LookupImpl(root, auth, useTls, requestTimeoutMs, topics); - this.functions = new FunctionsImpl(root, auth, asyncHttpConnector, requestTimeoutMs); - this.sources = new SourcesImpl(root, auth, asyncHttpConnector, requestTimeoutMs); - this.sinks = new SinksImpl(root, auth, asyncHttpConnector, requestTimeoutMs); - this.worker = new WorkerImpl(root, auth, requestTimeoutMs); - this.schemas = new SchemasImpl(root, auth, requestTimeoutMs); - this.bookies = new BookiesImpl(root, auth, requestTimeoutMs); - this.packages = new PackagesImpl(root, auth, asyncHttpConnector, requestTimeoutMs); - this.transactions = new TransactionsImpl(root, auth, requestTimeoutMs); - this.metadataMigration = new MetadataMigrationImpl(root, auth, requestTimeoutMs); - this.scalableTopics = new ScalableTopicsImpl(root, auth, requestTimeoutMs); + this.clusters = lendAuthExecutor(new ClustersImpl(root, auth, requestTimeoutMs)); + this.brokers = lendAuthExecutor(new BrokersImpl(root, auth, requestTimeoutMs)); + this.brokerStats = lendAuthExecutor(new BrokerStatsImpl(root, auth, requestTimeoutMs)); + this.proxyStats = lendAuthExecutor(new ProxyStatsImpl(root, auth, requestTimeoutMs)); + this.tenants = lendAuthExecutor(new TenantsImpl(root, auth, requestTimeoutMs)); + this.resourcegroups = lendAuthExecutor(new ResourceGroupsImpl(root, auth, requestTimeoutMs)); + this.namespaces = lendAuthExecutor(new NamespacesImpl(root, auth, requestTimeoutMs)); + this.topics = lendAuthExecutor(new TopicsImpl(root, auth, requestTimeoutMs)); + this.localTopicPolicies = lendAuthExecutor(new TopicPoliciesImpl(root, auth, requestTimeoutMs, false)); + this.globalTopicPolicies = lendAuthExecutor(new TopicPoliciesImpl(root, auth, requestTimeoutMs, true)); + this.nonPersistentTopics = lendAuthExecutor(new NonPersistentTopicsImpl(root, auth, requestTimeoutMs)); + this.resourceQuotas = lendAuthExecutor(new ResourceQuotasImpl(root, auth, requestTimeoutMs)); + this.lookups = lendAuthExecutor(new LookupImpl(root, auth, useTls, requestTimeoutMs, topics)); + this.functions = lendAuthExecutor(new FunctionsImpl(root, auth, asyncHttpConnector, requestTimeoutMs)); + this.sources = lendAuthExecutor(new SourcesImpl(root, auth, asyncHttpConnector, requestTimeoutMs)); + this.sinks = lendAuthExecutor(new SinksImpl(root, auth, asyncHttpConnector, requestTimeoutMs)); + this.worker = lendAuthExecutor(new WorkerImpl(root, auth, requestTimeoutMs)); + this.schemas = lendAuthExecutor(new SchemasImpl(root, auth, requestTimeoutMs)); + this.bookies = lendAuthExecutor(new BookiesImpl(root, auth, requestTimeoutMs)); + this.packages = lendAuthExecutor(new PackagesImpl(root, auth, asyncHttpConnector, requestTimeoutMs)); + this.transactions = lendAuthExecutor(new TransactionsImpl(root, auth, requestTimeoutMs)); + this.metadataMigration = lendAuthExecutor(new MetadataMigrationImpl(root, auth, requestTimeoutMs)); + this.scalableTopics = lendAuthExecutor(new ScalableTopicsImpl(root, auth, requestTimeoutMs)); if (originalCtxLoader != null) { Thread.currentThread().setContextClassLoader(originalCtxLoader); @@ -240,6 +243,8 @@ public PulsarAdminImpl(String serviceUrl, ClientConfigurationData clientConfigDa // trustCertsFilePath being the common case) leaks one of each. The connectors release only // their own borrowed handles, which is exactly why this has to happen here. asyncConnectorProvider.close(); + // A services-aware plugin's start() may already have run work here before the failure. + blockingAuthExecutor.shutdown(); } } } @@ -610,40 +615,53 @@ private void bindAuthenticationServices(ClientConfigurationData conf) { // does schedule some gets the framework's shared one — binding a scheduled pool per admin to sit idle // would cost more than it buys. // - // The blocking executor *is* the admin's own. This is where the SASL-over-HTTP challenge rounds put - // their GSSAPI work, so it must never be the admin's request threads — a slow KDC would consume them - // — but leaving it unbound is not right either: the plugin then shares one process-wide pool with - // every other client in the JVM, so a stalled identity provider reached by one admin throttles - // authentication for all of them. A small bounded pool per admin keeps that blast radius inside the - // admin that owns the plugin, and is shut down with it. + // The blocking executor *is* the admin's own — the same pool BaseResource runs the deprecated v4 + // composition on. This is where the SASL-over-HTTP challenge rounds put their GSSAPI work, so it must + // never be the admin's request threads — a slow KDC would consume them — but leaving it unbound is + // not right either: the plugin then shares one process-wide pool with every other client in the JVM, + // so a stalled identity provider reached by one admin throttles authentication for all of them. A + // small bounded pool per admin keeps that blast radius inside the admin that owns the plugin, and is + // shut down with it. ClientAuthenticationServices services = new DefaultClientAuthenticationServices( - authHttpClientFactory, null, blockingAuthExecutor(), Clock.systemDefaultZone(), openTelemetry, + authHttpClientFactory, null, blockingAuthExecutor, Clock.systemDefaultZone(), openTelemetry, clientInstanceId); aware.bindClientAuthenticationServices(services); } /** - * The admin's bounded executor for potentially-blocking authentication work (PIP-478). + * Lend a resource this admin's blocking authentication executor, so the deprecated v4 credential + * composition it runs per request stays on the admin's own pool (PIP-478). + * + * @param resource the freshly constructed resource + * @param the resource type + * @return the same resource + */ + private T lendAuthExecutor(T resource) { + resource.setBlockingAuthExecutor(blockingAuthExecutor); + return resource; + } + + /** + * Build the admin's bounded executor for potentially-blocking authentication work (PIP-478). * - *

Queues rather than rejects: every caller is an authenticated request, so a saturated pool must slow - * requests down rather than fail them. Core threads time out, so an admin whose plugin never blocks pays - * for nothing. Created lazily under this admin's monitor and shut down in {@link #close()}. + *

Queues rather than rejects, matching the framework's shared pool: every caller is an authenticated + * request, so a saturated pool must slow requests down rather than fail them — and a queued task is far + * smaller than what the caller already retains to produce it (a synchronous admin call is a parked + * thread, bounded by its own {@code requestTimeoutMs}). Core threads time out, so an admin whose plugin + * never blocks holds no threads. Shut down in {@link #close()}. * * @return the blocking authentication executor */ - private synchronized Executor blockingAuthExecutor() { - if (blockingAuthExecutor == null) { - ThreadPoolExecutor executor = new ThreadPoolExecutor(AUTH_BLOCKING_MAX_THREADS, - AUTH_BLOCKING_MAX_THREADS, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), - runnable -> { - Thread thread = new Thread(runnable, "pulsar-admin-auth-blocking"); - thread.setDaemon(true); - return thread; - }); - executor.allowCoreThreadTimeOut(true); - blockingAuthExecutor = executor; - } - return blockingAuthExecutor; + private static ThreadPoolExecutor newBlockingAuthExecutor() { + ThreadPoolExecutor executor = new ThreadPoolExecutor(AUTH_BLOCKING_MAX_THREADS, + AUTH_BLOCKING_MAX_THREADS, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), + runnable -> { + Thread thread = new Thread(runnable, "pulsar-admin-auth-blocking"); + thread.setDaemon(true); + return thread; + }); + executor.allowCoreThreadTimeOut(true); + return executor; } @Override @@ -667,10 +685,7 @@ public void close() { // PIP-478: after auth.close(), so a plugin shutting down over this executor still has it. shutdown() // rather than shutdownNow(): queued work is a credential call for a request already in flight, and // the threads are daemons, so a straggler cannot hold the JVM up. - ThreadPoolExecutor authExecutor = blockingAuthExecutor; - if (authExecutor != null) { - authExecutor.shutdown(); - } + blockingAuthExecutor.shutdown(); } @VisibleForTesting diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java index 9632c9cb8fcc8..37366e81075f2 100644 --- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java @@ -23,8 +23,11 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.Authentication; import org.apache.pulsar.client.api.AuthenticationDataProvider; import org.apache.pulsar.client.api.PulsarClientException; @@ -84,6 +87,36 @@ public void close() { } } + /** + * A v4 plugin of the multi-round shape: it completes the authentication stage from a thread of its own, + * the way a challenge-response plugin completes it from its HTTP callback thread, and records where the + * second hook — {@code newRequestHeader} — then ran. + */ + private static class AsyncStageAuthentication extends ThreadRecordingAuthentication { + + private final CompletableFuture>> stage = new CompletableFuture<>(); + private final AtomicReference newRequestHeaderThread = new AtomicReference<>(); + + @Override + public void authenticationStage(String requestUrl, AuthenticationDataProvider authData, + Map previousResHeaders, + CompletableFuture> authFuture) { + // Hand the stage back to the test rather than completing it here, so the continuation is + // registered before the completion happens. Completing it from a thread started here would race: + // an already-completed stage runs a plain continuation on the *registering* thread, which is the + // blocking executor, and the assertion below would then hold whether or not the hop exists. + stage.complete(authFuture); + } + + @Override + public Set> newRequestHeader(String hostName, + AuthenticationDataProvider authData, + Map previousResHeaders) { + newRequestHeaderThread.set(Thread.currentThread()); + return Set.of(Map.entry("X-Test", "value")); + } + } + /** The minimal concrete resource needed to reach the protected header composition. */ private static class TestResource extends BaseResource { TestResource(Authentication auth) { @@ -131,4 +164,66 @@ public boolean hasDataForHttp() { assertThat(headers).isNull(); } + + /** + * The composition belongs on the pool of the admin that owns the plugin, not on the framework's + * process-wide fallback: a stalled identity provider reached by one admin must not occupy the threads + * every other client in the JVM shares. Built through a real {@link PulsarAdminImpl} so what is pinned + * is that the admin actually lends its pool to the resources it constructs, not merely that + * {@link BaseResource} would use one if given it. + */ + @Test + public void theV4CompositionRunsOnTheOwningAdminsOwnPool() throws Exception { + ThreadRecordingAuthentication auth = new ThreadRecordingAuthentication(); + try (PulsarAdmin admin = PulsarAdmin.builder() + .serviceHttpUrl("http://broker.example:8080") + .authentication(auth) + .build()) { + // Same package, so the protected composition is reachable on the admin's own resource. No request + // is issued: computeAuthHeaders is the pre-request stage. + BaseResource resource = (BaseResource) admin.clusters(); + resource.computeAuthHeaders(URI.create("http://broker.example:8080/admin/v2/clusters")) + .get(30, TimeUnit.SECONDS); + + assertThat(auth.getAuthDataThread.get()).isNotNull(); + assertThat(auth.getAuthDataThread.get().getName()) + .as("the v4 credential must resolve on the admin's own bounded pool") + .startsWith("pulsar-admin-auth-blocking"); + } + } + + /** + * {@code newRequestHeader} is the second synchronous v4 hook. A plugin that completes the stage + * asynchronously completes it from a thread of its own — its HTTP callback thread — and a plain + * continuation would run the hook there, so off-loading the credential resolution alone would leave half + * the composition where PIP-478 says it must not be. + */ + @Test + public void theNewRequestHeaderHookIsOffLoadedToo() throws Exception { + AsyncStageAuthentication auth = new AsyncStageAuthentication(); + ExecutorService lentPool = Executors.newSingleThreadExecutor( + runnable -> new Thread(runnable, "lent-auth-pool")); + try { + TestResource resource = new TestResource(auth); + resource.setBlockingAuthExecutor(lentPool); + + CompletableFuture> composed = + resource.headers(URI.create("http://broker.example:8080/admin/v2/clusters")); + + // The plugin's own thread completes the stage, standing in for its HTTP callback thread. + CompletableFuture> stage = auth.stage.get(30, TimeUnit.SECONDS); + Thread completer = new Thread(() -> stage.complete(Map.of()), "stage-completer"); + completer.start(); + completer.join(); + + assertThat(composed.get(30, TimeUnit.SECONDS)).containsExactly(Map.entry("X-Test", "value")); + assertThat(auth.newRequestHeaderThread.get()) + .as("the second v4 hook must not run on the thread the plugin completed the stage from") + .isNotNull() + .isNotSameAs(completer); + assertThat(auth.newRequestHeaderThread.get().getName()).isEqualTo("lent-auth-pool"); + } finally { + lentPool.shutdownNow(); + } + } } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java index 70983fc30532a..6b72a18cd1ffc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java @@ -387,7 +387,12 @@ private CompletableFuture> v4AuthHeaders(URI uri) { // and newRequestHeader returns their static HTTP headers. CompletableFuture> stage = new CompletableFuture<>(); authentication.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 completes it from its own HTTP callback thread — so a + // plain continuation would run that hook there, which off-loading the resolution alone does not + // cover. Where the stage completes inline (the single-pass default) the hop is to a sibling task + // on this same executor, and where no executor was supplied it is a direct call as before. + return stage.thenApplyAsync(respHeaders -> { try { Set> headers = authentication.newRequestHeader(uri.toString(), authData, respHeaders); @@ -400,7 +405,7 @@ private CompletableFuture> v4AuthHeaders(URI uri) { } catch (Exception e) { throw new CompletionException(e); } - }); + }, blockingAuthExecutor); } catch (Throwable t) { return CompletableFuture.failedFuture(t); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java index 96a32278d8080..23526d3dfc38a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5AuthContexts.java @@ -81,6 +81,21 @@ static Executor sharedBlockingExecutor() { return SharedBlockingExecutor.INSTANCE; } + /** + * Resolve the executor that potentially-blocking authentication work must run on: the one the owning + * component lent, or the shared fallback pool when it lent none. Never the caller thread. + * + *

Exposed for a caller that has to run more than one step on the same executor — the deprecated v4 + * HTTP composition runs the credential resolution and the {@code newRequestHeader} continuation there — + * and so needs the resolved instance rather than only {@link #supplyBlocking}'s internal choice. + * + * @param blockingExecutor the bound blocking executor, or {@code null} if none was bound + * @return the executor to run the blocking work on; never {@code null} + */ + public static Executor blockingExecutorOrShared(Executor blockingExecutor) { + return blockingExecutor != null ? blockingExecutor : sharedBlockingExecutor(); + } + /** * @param brokerHost the broker host * @return a new binary-protocol call context with a fresh state slot @@ -109,7 +124,7 @@ public static AuthenticationCallContext binaryCallContext(String brokerHost) { * @return a future of the result; never throws synchronously */ public static CompletableFuture supplyBlocking(Executor blockingExecutor, Supplier task) { - Executor executor = blockingExecutor != null ? blockingExecutor : sharedBlockingExecutor(); + Executor executor = blockingExecutorOrShared(blockingExecutor); try { return CompletableFuture.supplyAsync(task, executor); } catch (Throwable t) { From 51f02671db3f19deb2ea640cd86ab0b532c602c9 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Fri, 14 Aug 2026 18:43:16 +0300 Subject: [PATCH 3/4] [test][admin] PIP-478: fail the build when a new admin resource is not lent the auth pool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on #26327 from david-streamlio. lendAuthExecutor(...) is opt-in per construction site. A resource added to PulsarAdminImpl later without the wrapper compiles, runs, and silently falls back to the framework's shared pool — the fix undone with no symptom, because the fallback works. Pinning one resource cannot catch that: the twenty-fourth is the one that would be missed. everyResourceTheAdminExposesIsLentThePool walks the accessors PulsarAdmin itself declares and asserts each resource reports the admin's pool, so the guard grows with the interface rather than with a hand-maintained list. It reflects over public accessors only — no reading of private state — and reads the executor through package-private accessors on both classes. Mutation-verified: dropping the wrapper from one construction site fails the test naming that accessor ("scalableTopics"). Assisted-by: Claude Code (Opus 5) --- .../client/admin/internal/BaseResource.java | 12 ++++++ .../admin/internal/PulsarAdminImpl.java | 11 +++++ .../admin/internal/AdminAuthOffloadTest.java | 43 +++++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java index 9296dcef07ea2..e1932e7bc7523 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/BaseResource.java @@ -95,6 +95,18 @@ 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(); diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java index 3f195210d5b53..79dbe84cf86b8 100644 --- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java +++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PulsarAdminImpl.java @@ -641,6 +641,17 @@ private T lendAuthExecutor(T resource) { return resource; } + /** + * The pool the resources above are lent, so a test can assert every one of them got it + * (VisibleForTesting). Lending is per construction site: a resource added later without the wrapper + * compiles, runs, and silently falls back to the shared pool. + * + * @return this admin's blocking authentication executor + */ + ThreadPoolExecutor blockingAuthExecutorForTest() { + return blockingAuthExecutor; + } + /** * Build the admin's bounded executor for potentially-blocking authentication work (PIP-478). * diff --git a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java index 37366e81075f2..733cbf290a08e 100644 --- a/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java @@ -19,10 +19,15 @@ package org.apache.pulsar.client.admin.internal; import static org.assertj.core.api.Assertions.assertThat; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; import java.net.URI; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -192,6 +197,44 @@ public void theV4CompositionRunsOnTheOwningAdminsOwnPool() throws Exception { } } + /** + * Lending is opt-in per construction site, so a resource added to {@link PulsarAdminImpl} later without + * the wrapper would compile, run, and silently fall back to the shared pool — the fix undone with no + * symptom. Walking the admin's own accessors is what turns that into a test failure; pinning one + * resource would not, since the twenty-fourth is the one that would be missed. + */ + @Test + public void everyResourceTheAdminExposesIsLentThePool() throws Exception { + try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl("http://broker.example:8080").build()) { + Executor pool = ((PulsarAdminImpl) admin).blockingAuthExecutorForTest(); + List notLent = new ArrayList<>(); + int checked = 0; + + for (Method accessor : PulsarAdmin.class.getMethods()) { + if (accessor.getParameterCount() != 0 || Modifier.isStatic(accessor.getModifiers()) + || accessor.getReturnType() == void.class) { + continue; + } + if (accessor.invoke(admin) instanceof BaseResource resource) { + checked++; + if (resource.blockingAuthExecutorForTest() != pool) { + notLent.add(accessor.getName()); + } + } + } + // The one resource accessor that takes an argument, so the loop above cannot reach it. + checked++; + if (((BaseResource) admin.topicPolicies(true)).blockingAuthExecutorForTest() != pool) { + notLent.add("topicPolicies(true)"); + } + + assertThat(notLent).as("admin resources left on the shared pool").isEmpty(); + assertThat(checked) + .as("the walk must actually reach the admin's resources, not silently filter them all out") + .isGreaterThanOrEqualTo(20); + } + } + /** * {@code newRequestHeader} is the second synchronous v4 hook. A plugin that completes the stage * asynchronously completes it from a thread of its own — its HTTP callback thread — and a plain From 5d3913e6dbc641be3eb3e21c612819b79178f504 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Thu, 20 Aug 2026 03:19:04 +0300 Subject: [PATCH 4/4] [improve][pip] PIP-478: record that the admin's v4 credential path is off-loaded The Error-model section still named BaseResource as "the exception" that "composes v4 headers on the calling thread" and said it was "tracked as a follow-up". This PR is that follow-up: BaseResource.computeAuthHeaders now runs the v4 composition on the bounded authentication executor the owning PulsarAdmin lends each resource, falling back to the shared pool for a resource built outside one. Leaving a closed follow-up recorded as open understates what shipped and makes the surrounding normative sentence read as false when it is now true. Assisted-by: Claude Code (Opus 5) --- pip/pip-478.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-478.md b/pip/pip-478.md index 2699fd3995dba..1ecc5ea72cc64 100644 --- a/pip/pip-478.md +++ b/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