[fix][proxy] PIP-478: resolve the proxy's broker-client credential off the event loop

pip/pip-478.md

@@ -142,7 +142,7 @@ This PIP introduces new API and implementation across the existing `pulsar-clien
 
 - **Non-Java client SDKs.** Each non-Java SDK (Python, Go, C++, Node.js) follows its own auth model and will be addressed by per-SDK PIPs.
 
-- **Off-loading the proxy's own broker-client credential I/O (known gap).** Motivation #1's event-loop-safety guarantee is delivered for `PulsarClient` / `PulsarAdmin` (and the broker's outbound clients, which are genuine clients that bind the framework's `ClientAuthenticationServices`). The **proxy's** connection to the broker is not a `PulsarClientImpl`: its lookup path uses a bare `ConnectionPool` and its data path a hand-rolled Netty `DirectProxyHandler`, and neither binds the client auth services (bounded blocking executor, framework HTTP client factory). On the proxy's **lookup** path the credential no longer runs on that loop: `ClientCnx` resolves and memoizes a binary authentication driver for a configuration that no `PulsarClient` owns, and with no services bound the credential call falls back to the framework's process-wide shared blocking pool. What remains is the **data** path: `DirectProxyHandler` resolves the credential inline on the proxy's Netty loop, on connect and on the broker's refresh sentinel — **the same behavior as v4, not a regression introduced by this PIP**. Closing it requires an async rework of that inline `getAuthData()`; the lookup leg additionally still runs without the framework HTTP client factory and on a shared library-owned pool rather than a client-owned bounded executor, which is why reordering proxy startup (the broker-client `Authentication` is created and started before the proxy's event loop, DNS resolver, and TLS factory exist) is still part of the follow-up. Both are deferred rather than bundled into this change.
+- **Binding the proxy's broker-client legs to client-owned services (known gap).** Motivation #1's event-loop-safety guarantee is delivered for `PulsarClient` / `PulsarAdmin` (and the broker's outbound clients, which are genuine clients that bind the framework's `ClientAuthenticationServices`). The **proxy's** connection to the broker is not a `PulsarClientImpl`: its lookup path uses a bare `ConnectionPool` and its data path a hand-rolled Netty `DirectProxyHandler`, and neither binds the client auth services (bounded blocking executor, framework HTTP client factory). On the proxy's **lookup** path the credential no longer runs on that loop: `ClientCnx` resolves and memoizes a binary authentication driver for a configuration that no `PulsarClient` owns, and with no services bound the credential call falls back to the framework's process-wide shared blocking pool. The **data** path is off-loaded too: `DirectProxyHandler` opens a per-connection exchange on a `ProxyService`-owned `V5BinaryAuthenticationDriver` and resolves the credential through it — on connect and on the broker's refresh sentinel alike — dispatching the continuation back to the channel's event loop, with challenge rounds strictly serialized (a non-refresh challenge arriving while a round is in flight is dropped) and bounded by the same round cap `ClientCnx` uses. What remains deferred: neither proxy leg binds the framework HTTP client factory, both run credential work on the shared library-owned pool rather than a client-owned bounded executor, and reordering proxy startup (the broker-client `Authentication` is created and started before the proxy's event loop, DNS resolver, and TLS factory exist) is still follow-up work.
 
 - **The broader FIPS-compliance mode.** This PIP covers the **TLS-transport** requirements for FIPS: a configurable TLS engine (JDK, not native BoringSSL) wired through every component plus the two configurable provider axes (Motivation #4, Goal #5). It does **not** define a full FIPS-mode profile: FIPS-approved algorithms in message encryption (key-wrap) and authentication (password hashing, token signing), a FIPS distribution/packaging variant (shipping `bc-fips` and excluding non-validated `bcprov` / `netty-tcnative-boringssl` / Conscrypt), and a fail-fast `fipsMode` validation switch are a **separate effort** — Pulsar-wide in scope and independent of this SPI. Concretely, the shipped `pulsar-server` distribution today bundles the **non-FIPS** BouncyCastle provider (`bcprov-jdk18on` / `bcpkix-jdk18on`) and explicitly excludes `bc-fips`, and it ships no `bctls-fips` (the jar registering `BCJSSE`) at all; because the two BouncyCastle families declare the same `org.bouncycastle.*` classes under different signers they cannot coexist on one classpath, so until that packaging effort lands a FIPS deployment assembles the provider classpath itself. The in-tree `pulsar-client-test-bcfips` module assembles a classpath that way — excluding the non-FIPS BouncyCastle jars in favour of `bc-fips` — but it covers the crypto side only: it ships no `bctls-fips` and sets neither provider key, so it is not an end-to-end FIPS TLS test. This PIP deliberately provides only the TLS-transport configurability those deployments require, so the two efforts compose without one blocking the other.
 

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/DirectProxyHandler.java

@@ -42,14 +42,15 @@
 import io.netty.util.CharsetUtil;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import lombok.CustomLog;
 import lombok.Getter;
 import lombok.SneakyThrows;
 import org.apache.pulsar.PulsarVersion;
 import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.AuthenticationDataProvider;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.api.AuthData;
 import org.apache.pulsar.common.api.proto.BaseCommand;
@@ -61,6 +62,7 @@
 import org.apache.pulsar.common.protocol.PulsarDecoder;
 import org.apache.pulsar.common.stats.Rate;
 import org.apache.pulsar.common.tls.impl.TlsContextAcquisition;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.common.util.netty.NettyChannelUtil;
 
 @CustomLog
@@ -79,8 +81,13 @@ public class DirectProxyHandler {
     private final String clientAuthMethod;
     public static final String TLS_HANDLER = "tls";
 
+    // PIP-478: hard cap on broker challenge rounds within a single binary authentication exchange, mirroring
+    // ClientCnx.MAX_AUTH_CHALLENGE_ROUNDS (both in turn mirror HttpAuthenticationDriver.MAX_CHALLENGE_ROUNDS).
+    // A broker that answers every CommandAuthResponse with another challenge would otherwise loop against the
+    // proxy forever, and each round now also schedules credential work onto a blocking pool.
+    static final int MAX_AUTH_CHALLENGE_ROUNDS = 10;
+
     private final Authentication authentication;
-    private AuthenticationDataProvider authenticationDataProvider;
     private final ProxyService service;
     private final Runnable onHandshakeCompleteAction;
     final boolean tlsEnabledWithBroker;
@@ -259,6 +266,18 @@ public class ProxyBackendHandler extends PulsarDecoder {
         private final ProxyConfiguration config;
         private final int protocolVersion;
         private final FeatureFlags featureFlags;
+        // PIP-478: the v5 exchange this backend connection authenticates through. Replaced on a broker-pushed
+        // REFRESH, which starts a fresh exchange. Only ever touched on this channel's event loop.
+        private AuthenticationExchange authExchange;
+        // PIP-478: round state. AuthenticationExchange is single-round and non-thread-safe, and serializing
+        // its rounds is the caller's obligation; this class is its second caller after ClientCnx. While the
+        // frame decoder is still running (state Init) two challenge frames arriving in one read reach
+        // handleAuthChallenge in the same event-loop turn, before either resolution has completed, and would
+        // otherwise drive the same exchange concurrently. Both fields are touched only on this channel's
+        // event loop (channelActive, handleAuthChallenge, and the continuations dispatched there), so they
+        // need no synchronization.
+        private boolean authRoundInProgress;
+        private int authChallengeRounds;
 
         public ProxyBackendHandler(ProxyConfiguration config, int protocolVersion, String remoteHostName,
                                    FeatureFlags featureFlags) {
@@ -275,16 +294,76 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
             if (config.isHaProxyProtocolEnabled()) {
                 writeHAProxyMessage();
             }
-
-            // Send the Connect command to broker
-            authenticationDataProvider = authentication.getAuthData(remoteHostName);
-            AuthData authData = authenticationDataProvider.authenticate(AuthData.INIT_AUTH_DATA);
-            ByteBuf command = Commands.newConnect(
-                    authentication.getAuthMethodName(), authData, protocolVersion,
-                    proxyConnection.clientVersion, null /* target broker */,
-                    originalPrincipal, clientAuthData, clientAuthMethod, PulsarVersion.getVersion(), featureFlags);
-            writeAndFlush(command);
             isTlsOutboundChannel = ProxyConnection.isTlsChannel(inboundChannel);
+
+            // Send the Connect command to broker. PIP-478: the credential is resolved through a v5 exchange
+            // rather than by calling the v4 plugin here. This method runs on the Netty event loop, and the v4
+            // call it used to make is arbitrary plugin code — an OAuth2 token endpoint round trip, an Athenz
+            // ZTS fetch, a GSSAPI exchange with the KDC — which stalled every connection multiplexed onto
+            // that loop for its duration. The exchange's calls always off-load.
+            //
+            // The auth method name is read from the v4 plugin here and in handleAuthChallenge below, where
+            // ClientCnx instead takes it from the exchange that produced the credential. Both are correct for
+            // the proxy — it owns one started plugin, and the bridge's authMethodName() delegates straight to
+            // it — and reading it from the plugin does not depend on a round having completed. Deliberate.
+            authExchange = service.getProxyClientAuthenticationDriver().newAuthenticationExchange(remoteHostName);
+            sendWhenResolved(authExchange.getAuthDataAsync(),
+                    authData -> Commands.newConnect(
+                            authentication.getAuthMethodName(), authData, protocolVersion,
+                            proxyConnection.clientVersion, null /* target broker */,
+                            originalPrincipal, clientAuthData, clientAuthMethod, PulsarVersion.getVersion(),
+                            featureFlags),
+                    "connect");
+        }
+
+        /**
+         * Send a command built from an asynchronously-resolved credential (PIP-478).
+         *
+         * <p>The continuation is dispatched onto this channel's event loop, so the command is built and
+         * written there whether the credential was already in memory or needed I/O — command ordering on the
+         * channel is therefore unchanged from the synchronous version. A failure closes the backend channel:
+         * the proxy has no credential to send, and leaving the connection open would wait out the broker's
+         * timeout instead of letting the client retry. That covers a failure to build the command too, not
+         * just a failure to resolve the credential.
+         *
+         * <p>This is where an authentication round begins: the round is marked in progress so that a
+         * challenge arriving before it completes is dropped rather than re-entering the exchange
+         * concurrently.
+         *
+         * @param resolution     the credential being resolved
+         * @param commandBuilder builds the command to send from the resolved credential
+         * @param what           what is being authenticated, for logging
+         */
+        private void sendWhenResolved(CompletableFuture<AuthData> resolution,
+                                      Function<AuthData, ByteBuf> commandBuilder, String what) {
+            authRoundInProgress = true;
+            // The future returned by whenCompleteAsync is intentionally discarded: the continuation handles
+            // every outcome itself, and the only way that future fails is ctx.executor() rejecting during
+            // event-loop shutdown — at which point the channel is already going away.
+            resolution.whenCompleteAsync((authData, throwable) -> {
+                authRoundInProgress = false;
+                if (throwable != null) {
+                    Throwable cause = FutureUtil.unwrapCompletionException(throwable);
+                    log.error().attr("channel", ctx.channel()).attr("stage", what).exception(cause)
+                            .log("Failed to resolve the proxy's broker-client credential");
+                    ctx.close();
+                    return;
+                }
+                if (!ctx.channel().isActive()) {
+                    // The backend connection went away while the credential was resolving. Logged so that
+                    // "backend connected but never sent CommandConnect" is diagnosable rather than silent.
+                    log.debug().attr("channel", ctx.channel()).attr("stage", what)
+                            .log("Backend channel closed while the proxy's broker-client credential resolved");
+                    return;
+                }
+                try {
+                    writeAndFlush(commandBuilder.apply(authData));
+                } catch (Throwable t) {
+                    log.error().attr("channel", ctx.channel()).attr("stage", what).exception(t)
+                            .log("Failed to send the proxy's broker-client authentication command");
+                    ctx.close();
+                }
+            }, ctx.executor());
         }
 
         @Override
@@ -345,37 +424,68 @@ protected void handleAuthChallenge(CommandAuthChallenge authChallenge) {
             checkArgument(authChallenge.hasChallenge());
             checkArgument(authChallenge.getChallenge().hasAuthData() && authChallenge.getChallenge().hasAuthData());
 
-            if (Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData())) {
-                try {
-                    authenticationDataProvider = authentication.getAuthData(remoteHostName);
-                } catch (PulsarClientException e) {
-                    log.error().attr("channel", ctx.channel())
-                            .exception(e)
-                            .log("Error refreshing authentication data provider");
-                    return;
+            // PIP-478 binary routing rule 2: the broker's REFRESH sentinel restarts authentication with a
+            // fresh exchange whose getAuthDataAsync() re-produces the current credential, rather than being
+            // routed into the conversation it just terminated. Any other challenge is a round of the current
+            // exchange, whose state slot carries conversation state across rounds. This mirrors ClientCnx.
+            // The REFRESH branch is conformance with that rule rather than a path the broker can reach here:
+            // this handler stops decoding once state == HandshakeCompleted, and the broker arms its refresh
+            // task with an initial delay of authenticationRefreshCheckSeconds after connect completes, so by
+            // the time a REFRESH is pushed it is proxied straight through to the client, which answers it.
+            // The proxy's own credential refresh lives in ProxyConnection, not here.
+            boolean refresh =
+                    Arrays.equals(AuthData.REFRESH_AUTH_DATA_BYTES, authChallenge.getChallenge().getAuthData());
+            // PIP-478 (serialize-or-drop): a broker never pipelines challenges — it waits for each
+            // CommandAuthResponse — so a challenge arriving while a round is still in flight is anomalous,
+            // and servicing it would re-enter the same single-round, non-thread-safe exchange concurrently.
+            // Dropping it makes rounds strictly serialized, which is why this class needs none of the
+            // generation guarding ClientCnx carries: nothing here can supersede an in-flight round. ClientCnx
+            // does need it, because it lets a REFRESH supersede one; the proxy can drop a REFRESH instead,
+            // both because it cannot reach this handler and because the broker's refresh check is a
+            // scheduleAtFixedRate task that would re-send it on the next tick.
+            if (authRoundInProgress) {
+                log.debug().attr("channel", ctx.channel())
+                        .log("Dropping a broker auth challenge received while an auth round is in progress");
+                return;
+            }
+            // Bound the exchange. A REFRESH opens a fresh exchange, so it resets the counter; any other
+            // challenge counts towards the cap.
+            if (refresh) {
+                authChallengeRounds = 0;
+            } else if (++authChallengeRounds > MAX_AUTH_CHALLENGE_ROUNDS) {
+                log.error().attr("channel", ctx.channel()).attr("maxChallengeRounds", MAX_AUTH_CHALLENGE_ROUNDS)
+                        .log("Binary authentication exceeded the maximum challenge rounds; closing the "
+                                + "broker connection");
+                ctx.close();
+                return;
+            }
+            CompletableFuture<AuthData> resolution;
+            try {
+                if (refresh) {
+                    authExchange =
+                            service.getProxyClientAuthenticationDriver().newAuthenticationExchange(remoteHostName);
+                    resolution = authExchange.getAuthDataAsync();
+                } else {
+                    resolution = authExchange
+                            .authenticateAsync(AuthData.of(authChallenge.getChallenge().getAuthData()));
                 }
+            } catch (Throwable t) {
+                // One try/catch so a plugin that throws synchronously fails the connection rather than
+                // propagating up the event loop.
+                resolution = CompletableFuture.failedFuture(t);
             }
 
             // mutual authn. If auth not complete, continue auth; if auth complete, complete connectionFuture.
-            try {
-                AuthData authData = authenticationDataProvider
-                    .authenticate(AuthData.of(authChallenge.getChallenge().getAuthData()));
-
+            sendWhenResolved(resolution, authData -> {
                 checkState(!authData.isComplete());
-
-                ByteBuf request = Commands.newAuthResponse(authentication.getAuthMethodName(),
-                    authData,
-                    this.protocolVersion,
-                    PulsarVersion.getVersion());
-
                 log.debug().attr("channel", ctx.channel())
                         .attr("authMethod", authentication.getAuthMethodName())
                         .log("Mutual auth");
-
-                writeAndFlush(request);
-            } catch (Exception e) {
-                log.error().exception(e).log("Error mutual verify");
-            }
+                return Commands.newAuthResponse(authentication.getAuthMethodName(),
+                        authData,
+                        this.protocolVersion,
+                        PulsarVersion.getVersion());
+            }, "challenge");
         }
 
         @Override