(pulsar) 03/05: [fix][client] PIP-478: give a bridged v4 plugin an executor when no client owns it

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch lh-pip-478-v5-native-auth-v2
in repository https://gitbox.apache.org/repos/asf/pulsar.git

commit 3ea0bdbe1a56c8a121f07b2b3671571f93fac9be
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Wed Aug 12 21:17:40 2026 +0300

    [fix][client] PIP-478: give a bridged v4 plugin an executor when no client owns it
    
    Driving the v5 model natively means every legacy v4 plugin now goes through
    LegacyV4AuthenticationAdapter, which by design refuses to run a v4 call inline — the
    caller thread is a Netty event loop, and running credential I/O there is the stall
    PIP-478 exists to remove. It off-loads to the executor its init context supplies.
    
    Not every connection is opened by a PulsarClient, though. The proxy builds broker
    connections straight from a ClientConfigurationData, and so do embedders and tests, so
    no client-owned executor is bound and the context supplied none: the adapter then failed
    every call with "used before initializeAsync() completed". That refuses connections v4
    made fine (ProxyAuthenticationTest, ProxySaslAuthenticationTest), and it only surfaced on
    the proxy because the built-in bodies never off-load.
    
    Supply a shared blocking executor for those callers instead. Running inline was the other
    option and is the wrong one: on that path the caller thread is an event loop too.
    The pool costs nothing when unused — no core threads, daemon threads retiring after a
    minute idle — and is created on first use.
    
    Verified by mutation: reverting this fails ProxySaslAuthenticationTest.testAuthentication
    with the same IllegalStateException CI reported.
    
    Assisted-by: Claude Code (Opus 5)
---
 .../pulsar/client/impl/auth/v5/V5AuthContexts.java | 39 +++++++++++++++++++---
 1 file changed, 35 insertions(+), 4 deletions(-)

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 1e3088768332..4653267af74d 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
@@ -25,6 +25,9 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
 import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext;
@@ -41,9 +44,10 @@ import org.apache.pulsar.http.PulsarHttpClientFactory;
  * HTTP client factory, client instance id) are late-bound into the init context via
  * {@link ClientAuthenticationServices}
  * — so a credential-fetching body (OAuth2, Athenz) can off-load its blocking fetch onto the blocking
- * executor instead of running it on the Netty event loop. When no services are bound (a plugin used
- * outside a client), the context degrades to {@code null} services and the body falls back to inline
- * computation.
+ * executor instead of running it on the Netty event loop. When no services are bound (a connection opened
+ * outside a {@code PulsarClient} — the proxy's broker connections, embedders, tests), the context reports
+ * no HTTP client factory or scheduler, but still supplies a shared blocking executor: the offload
+ * discipline holds on every path, since the caller thread there is a Netty event loop too.
  */
 public final class V5AuthContexts {
 
@@ -103,6 +107,33 @@ public final class V5AuthContexts {
         }
     }
 
+    /**
+     * The blocking executor used when no client services are bound.
+     *
+     * <p>Not every connection is opened by a {@code PulsarClient}: the proxy builds broker connections
+     * straight from a configuration, and so do embedders and tests. Those callers have no client-owned
+     * executor to lend, but a bridged v4 plugin still must not run its credential call on the caller
+     * thread — which on that path is a Netty event loop. Falling back to running inline would reintroduce
+     * exactly the stall PIP-478 exists to remove, and refusing to run at all would break connections v4
+     * made fine, so the library keeps one shared pool for them.
+     *
+     * <p>It costs nothing when unused: no core threads, daemon threads that retire after a minute of idle.
+     * Held in a holder class so it is created on first use rather than on class load.
+     */
+    private static final class SharedBlockingExecutor {
+        static final Executor INSTANCE = create();
+
+        private static Executor create() {
+            ThreadPoolExecutor executor = new ThreadPoolExecutor(0, 8, 60L, TimeUnit.SECONDS,
+                    new SynchronousQueue<>(), runnable -> {
+                        Thread thread = new Thread(runnable, "pulsar-auth-blocking-shared");
+                        thread.setDaemon(true);
+                        return thread;
+                    });
+            return executor;
+        }
+    }
+
     private static final class InitContext implements AuthenticationInitContext {
         private final String clientInstanceId;
 
@@ -122,7 +153,7 @@ public final class V5AuthContexts {
 
         @Override
         public Executor blockingExecutor() {
-            return null;
+            return SharedBlockingExecutor.INSTANCE;
         }
 
         @Override