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 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..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 @@ -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 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,13 +169,50 @@ 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. + // + // 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> v4AuthHeaders(URI uri, Executor blockingExecutor) { + try { AuthenticationDataProvider authData = auth.getAuthData(uri.getHost()); if (!authData.hasDataForHttp()) { return CompletableFuture.completedFuture(null); } 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) { @@ -155,7 +224,7 @@ protected CompletableFuture> computeAuthHeaders(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 473ab781e9da8..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 @@ -29,6 +29,8 @@ import java.time.Clock; import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import lombok.CustomLog; import lombok.Getter; @@ -90,6 +92,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 +130,12 @@ 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. 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; @@ -155,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); @@ -193,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); @@ -229,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(); } } } @@ -595,16 +611,70 @@ 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 — 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, null, Clock.systemDefaultZone(), openTelemetry, clientInstanceId); + authHttpClientFactory, null, blockingAuthExecutor, Clock.systemDefaultZone(), openTelemetry, + clientInstanceId); aware.bindClientAuthenticationServices(services); } + /** + * 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; + } + + /** + * 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). + * + *

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 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 public void close() { try { @@ -623,6 +693,10 @@ 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. + 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 new file mode 100644 index 0000000000000..733cbf290a08e --- /dev/null +++ b/pulsar-client-admin/src/test/java/org/apache/pulsar/client/admin/internal/AdminAuthOffloadTest.java @@ -0,0 +1,272 @@ +/* + * 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.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; +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; +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() { + } + } + + /** + * 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) { + 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(); + } + + /** + * 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"); + } + } + + /** + * 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 + * 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 23baef8960485..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) { @@ -242,12 +257,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(); + } + } +}