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

[lhotari]

Main Issue: #25890

PIP: #25890

Stacked on #26327 — this PR's base is lh-pip-478-admin-auth-executor, which sits on #26326 → #26322 → #26319 → #26317 and thence on master. Review those first; the diff here shows only this part. A stacked PR runs only the semantic-title check until its base merges.

Motivation

DirectProxyHandler.channelActive called the v4 authentication plugin inline — getAuthData() then
authenticate(INIT_AUTH_DATA) — on the thread delivering the channel-active event, and
handleAuthChallenge did the same for every challenge round.

That thread is a Netty I/O loop, and the code being called is arbitrary plugin code: an OAuth2 token
endpoint round trip, an Athenz ZTS fetch, a GSSAPI exchange with the KDC. While it ran, every
connection multiplexed onto that loop stalled — the exact hazard PIP-478 exists to remove, and the one
it already removed from the client.

pip-478.md:145 put the proxy's broker-client credential I/O out of scope, on the reasoning that the
client still had a synchronous path anyway. After the v5-native inversion (#26317) the client has no
synchronous path at all, so there is no longer a justification for keeping one here — this is the last
place in the codebase where plugin credential code runs on an event loop.

Modifications

ProxyService owns one V5BinaryAuthenticationDriver, built lazily from its already-started v4
plugin (the bridge must not run that lifecycle a second time, since ProxyService starts and closes
that instance itself). Each DirectProxyHandler opens its own exchange against it — the exchange is
what carries per-connection conversation state, so a multi-round mechanism still works across rounds.
The getter reads the memoized driver without holding the monitor and synchronizes only on the miss, as
ClientCnx.resolveAuthDriver does: this is the ProxyService monitor, shared with the metrics-servlet
accessors, and every backend channelActive would otherwise take it.

The credential itself is unchanged. ProxyConnection.getClientAuthentication() returns
service.getProxyClientAuthenticationPlugin(), so the plugin this class used before and the plugin the
driver is built from are the same started instance. Routing through ProxyService changes where the
credential is resolved, not which credential is sent.

No ClientAuthenticationServices are bound: the proxy is not a PulsarClient and has no client-owned
executor to lend. Credential work therefore lands on the framework's shared blocking pool, which is
the case V5AuthContexts documents for exactly this caller — the alternative, running it inline, is
the event loop this change exists to free.

The exchange's rounds are serialized and bounded. AuthenticationExchange is single-round and
non-thread-safe, and its javadoc makes serializing rounds the caller's obligation; this class is its
second caller after ClientCnx. A challenge arriving while a round is still in flight is dropped
rather than re-entering the exchange concurrently — two challenge frames delivered in one read reach
handleAuthChallenge in the same event-loop turn, before either resolution has completed. Rounds are
then strictly serialized, which is why this class needs none of the generation guarding ClientCnx
carries: nothing here can supersede an in-flight round. A challenge round cap
(MAX_AUTH_CHALLENGE_ROUNDS, mirroring ClientCnx) closes the backend connection rather than letting
a broker that answers every CommandAuthResponse with another challenge loop forever — each round now
also schedules work onto a blocking pool.

Three properties are preserved deliberately:

• Command ordering is unchanged. The continuation is dispatched back onto the channel's own
  event loop, so the connect command and every auth response are still built and written there,
  whether the credential was already in memory or needed I/O.
• A resolution failure now closes the backend channel rather than only logging. The proxy has no
  credential to send; leaving the connection open would wait out the broker's timeout instead of
  letting the client retry. This mirrors what ClientCnx.completeAuthChallenge was fixed to do. The
  same applies to a failure to build the command: checkState(!authData.isComplete()) used to
  throw into a catch that logged and left the connection open, and now closes it too.
• The broker-pushed REFRESH sentinel starts a fresh exchange, per PIP-478 binary routing rule 2,
  mirroring ClientCnx — rather than being routed into the conversation it just terminated. That
  branch is conformance with the rule, not a path the broker can reach here: see Verifying below.

The auth method name is read from the v4 plugin rather than from the exchange (where ClientCnx reads
it). Both are correct here — the proxy owns one started plugin and the bridge's authMethodName()
delegates to it — and reading it from the plugin does not depend on a round having completed; a comment
now says so, since next to ClientCnx it reads like an oversight.

The now-unused AuthenticationDataProvider field is removed.

Verifying this change

DirectProxyHandlerAuthTest pins the properties that live inside ProxyBackendHandler, against an
EmbeddedChannel: its event loop runs a submitted task only when the test asks it to, which turns "did
this run on the event loop or off it" from a race into an assertion. All four are
mutation-verified — each fails, alone and for its own reason, when the guard it pins is removed:

• theConnectCredentialIsResolvedOffTheEventLoop / theChallengeRoundIsResolvedOffTheEventLoop —
  the v4 plugin records the thread of every credential call; both must land on
  pulsar-auth-blocking-shared. Restoring either inline call records the event-loop thread instead.
• aChallengeArrivingWhileARoundIsInFlightIsDropped — reproduces two challenges reaching the handler
  before the first resolution completes. Without the guard the second one re-enters the exchange.
• anEndlesslyChallengingBrokerIsCutOffAtTheRoundCap — 11 challenges; the exchange must answer 10 and
  the channel must close. Without the cap it answers all 11 and keeps going.

The existing proxy authentication suites cover the connect path end to end and still pass:
ProxyRefreshAuthTest, ProxyForwardAuthDataTest, ProxyAuthenticatedProducerConsumerTest,
ProxyWithAuthorizationTest, ProxyAuthenticationTest.

Not covered, because it is not reachable here: the broker-pushed REFRESH branch.
ProxyBackendHandler.channelRead stops decoding once state == HandshakeCompleted, and the broker
arms its refresh task (scheduleAtFixedRate) with an initial delay of
authenticationRefreshCheckSeconds after connect completes — so a REFRESH is always proxied straight
through to the client as raw bytes, and the client answers it. That is what ProxyRefreshAuthTest
asserts on: ClientCnx.getLastDisconnectedTimestamp(), the client's refresh through the proxy. The
proxy's own credential refresh lives in ProxyConnection (#25179), not in this class. An earlier
revision of this description claimed the suites covered the REFRESH path through the changed code;
they do not, and nothing can, so the branch is documented as routing-rule conformance instead.

./gradlew sanityCheck and quickCheck pass.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Threading model: the proxy's broker-client credential resolution moves off the Netty event loop
onto the framework's shared blocking pool. Command construction and writes stay on the event loop, so
ordering on the channel is unchanged.

[david-streamlio]

The motivation is right and the shape of the fix is the one I'd want: ProxyService owns one driver, each DirectProxyHandler opens its own exchange, and the continuation is dispatched back onto the channel's event loop so command construction and writes stay where they were. I traced the load-bearing invariants rather than taking them from the description, and they hold:

• Same credential is still sent. ProxyConnection.getClientAuthentication() returns service.getProxyClientAuthenticationPlugin(), so the authentication field this class used before and the plugin the new driver is built from are the same instance. Routing through ProxyService doesn't change which credential goes to the broker. This is the non-obvious premise of the whole refactor and is worth a word in the PR body.
• Per-host scoping and the initial call are preserved verbatim. newAuthenticationExchange(remoteHostName) → V5AuthContexts.binaryCallContext(brokerHost) → LegacyV4CredentialAdapter.getAuthDataAsync does v4.getAuthData(callContext.brokerHost()) then d.authenticate(AuthData.INIT_AUTH_DATA) — exactly what channelActive did inline.
• Challenge rounds keep the same provider. The adapter stashes the AuthenticationDataProvider in the call-context state slot, which reproduces the old retained authenticationDataProvider field for multi-round mechanisms. Rounds stay serialized here because the next challenge only arrives after the previous response is written.
• The lifecycle claim checks out. forStartedV4Plugin → wrapAlreadyStarted → ownsLifecycle=false, and the adapter gates both v4.start() in initializeAsync and v4.close() in close() on that flag.
• Ordering is genuinely unchanged. Inbound client bytes reach the backend only through ProxyConnection.channelRead, gated on this.directProxyHandler != null, which is assigned only in handleBrokerConnected. So nothing can be written to the backend channel between channelActive and the deferred connect write. The HAProxy header still goes out synchronously first, and moving isTlsOutboundChannel above the send is correct.

Findings, in the order I'd act on them:

1. The REFRESH path isn't covered by the cited tests — and isn't reachable in this class.

ProxyBackendHandler.channelRead stops decoding once state == HandshakeCompleted (set in handleConnected) and forwards raw bytes to the client. On the broker side, ServerCnx.maybeScheduleAuthenticationCredentialsRefresh only schedules the refresh task at authenticationRefreshCheckSeconds (default 60s) after connect completes. So the broker's REFRESH sentinel can never reach DirectProxyHandler.handleAuthChallenge — it arrives long after the handler stopped decoding, and gets proxied straight through to the client.

ProxyRefreshAuthTest asserts on ClientCnx.getLastDisconnectedTimestamp() over pulsarClientImpl.getCnxPool().getConnections() — it exercises the client's refresh through the proxy, not this code path.

So "the proxy authentication suites exercise both the connect path and the broker-pushed REFRESH path end to end through the changed code" holds for connect but not for REFRESH. Since the REFRESH branch is where the semantics actually changed (old: new provider + authenticate(REFRESH_AUTH_DATA); new: fresh exchange + getAuthDataAsync() → authenticate(INIT_AUTH_DATA), the delta pip-478.md:972 documents), I'd rather the PR body said plainly that the branch is defensive parity with ClientCnx/ProxyClientCnx and is not reachable here, than claimed test coverage for it. Not a code change — a claim change.

2. getProxyClientAuthenticationDriver() puts a shared monitor on the per-connection hot path. (ProxyService.java:702)

Every backend channelActive, on every proxy event loop, takes the ProxyService monitor — which is also held by createMetricsServlet() (:400), resetMetricsServlet() (:601), getMetricsServlet() (:710) and addPrometheusRawMetricsProvider() (:714). Those are all cheap, so this isn't a correctness bug, but the sibling code in this same stack — ClientCnx.resolveAuthDriver — deliberately does a lock-free read first and only synchronizes on the miss, precisely to keep the connect path off a shared monitor. Matching that here (make proxyClientAuthenticationDriver volatile, then double-check) seems worth it in a PR about not making connections wait on shared machinery. Note volatile isn't optional if you do this — the field is currently plain, so an unsynchronized read would be a data race.

3. The command-build failure path also closes now, which the PR body doesn't mention.

checkState(!authData.isComplete()) (:418) used to throw into catch (Exception e) { log.error("Error mutual verify") } and leave the connection open. It now lands in sendWhenResolved's builder catch and closes the channel. I think that's the right call and consistent with the stated policy, but the body attributes the new close behaviour only to resolution failures. Worth one more sentence so it isn't discovered later as a surprise.

4. Auth method name is read from the v4 plugin, not the exchange. (:292, :422)

ClientCnx takes the name from the exchange that produced the credential; this class still reads authentication.getAuthMethodName(). For the proxy the two agree (same plugin instance, and the bridge's authMethodName() delegates to v4.getAuthMethodName()), so this is correct as written — and arguably better here, since it doesn't depend on a completed round. But it reads as an oversight next to ClientCnx. A short comment saying why would stop someone "fixing" it later.

5. Two silent paths in sendWhenResolved. (:314, :322)

• The future returned by whenCompleteAsync is discarded. If ctx.executor() rejects during event-loop shutdown, that rejection completes the discarded future and vanishes — no log, no close. Harmless in practice since the channel is dying anyway, but a // intentionally discarded would say so on purpose.
• The !ctx.channel().isActive() early return leaves no trace. A log.debug() there would make "backend connected but never sent CommandConnect" diagnosable; right now that outcome is invisible.

On the test you removed. Rather than a fixture that has to authenticate end to end, the property is pinnable much more cheaply: configure the proxy with a small v4 plugin whose getAuthData records Thread.currentThread().getName(), then assert the recorded name is pulsar-auth-blocking-shared (or at least isn't the Netty loop). That sidesteps the TLS/fixture setup entirely and pins exactly the invariant this PR exists to establish. Happy to write it if useful.

Nothing here is a blocker on the code as written — 1 is a claim in the description, 2–5 are polish. The functional core looks right to me.

[Copilot]

Pull request overview

Moves proxy-to-broker credential resolution off Netty event loops using the v5 authentication driver.

Changes:

• Adds a shared, lazily initialized authentication driver.
• Resolves connect and challenge credentials asynchronously.
• Closes backend channels when credential resolution fails.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
ProxyService.java	Owns the shared v5 authentication driver.
DirectProxyHandler.java	Asynchronously handles broker authentication exchanges.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[david-streamlio]

Following up on the Copilot review that landed alongside mine, because one of its comments is correct in a way my review could be misread to dismiss.

Its comment on DirectProxyHandler.java:408 asks for round-state guarding equivalent to ClientCnx. I checked, and the parity gap is real:

Guard	ClientCnx	DirectProxyHandler
authRoundInProgress serialize-or-drop	yes	no
generation guard suppressing superseded continuations	yes	no
MAX_AUTH_CHALLENGE_ROUNDS cap	yes	no

ClientCnx.handleAuthChallenge states the reason itself: servicing a challenge that arrives mid-round "would re-enter the same single-round, non-thread-safe exchange concurrently". AuthenticationExchange's own javadoc makes serialization a caller obligation — "its rounds are serialized by the caller ({@code ClientCnx} issues the next round only after the previous future completes), so an implementation needs no internal synchronization". DirectProxyHandler is now a second caller of that contract and does not honour it.

Scoping this against what I wrote, since the two reviews interact:

My review argued the broker-pushed REFRESH sentinel cannot reach handleAuthChallenge here — channelRead stops decoding once state == HandshakeCompleted, and ServerCnx.maybeScheduleAuthenticationCredentialsRefresh only arms the refresh task authenticationRefreshCheckSeconds (default 60s) after connect. I still believe that, and it does rule out the specific REFRESH-supersedes-an-in-flight-round scenario in Copilot's comment.

It does not rule out the rest, and I should have flagged that and didn't. During Init the handler decodes every frame, so two challenge frames arriving in one TCP read produce two handleAuthChallenge invocations in the same event-loop turn — before either resolution future has completed — and both then drive the same non-thread-safe exchange. Nothing in this class prevents it. Please don't take my unreachability argument as an answer to the whole comment; it answers one third of it.

The missing round cap is the part I'd act on first. It is not a concurrency question at all: a broker that keeps answering each CommandAuthResponse with another challenge will loop indefinitely against the proxy, where ClientCnx fails the connection once MAX_AUTH_CHALLENGE_ROUNDS is exceeded. Each round now also schedules work onto the shared blocking pool. The proxy is the more exposed of the two components, so it is the one that least wants the unbounded version of this loop.

Copilot's other comment — that the event-loop property isn't pinned by a test — restates what the PR body already discloses, so nothing new there; my suggestion about a recording plugin still stands as the cheap way to get it.

[lhotari]

Thanks — and particularly for the follow-up that scoped the two reviews against each other rather than
leaving the unreachability argument to be read as an answer to all of Copilot's comment. That framing
is what the fix is built on. Everything below is in d0a4cfece48.

1. The REFRESH claim — corrected, and it is stronger than a timing argument

You are right, and the description was wrong. It is also not merely that the broker's refresh task is
armed too late: after the handshake the backend connection stops decoding altogether and switches to
byte-level proxying, so a REFRESH frame is structurally unable to reach handleAuthChallenge — it is
forwarded to the client, which answers it. That is what ProxyRefreshAuthTest asserts on
(ClientCnx.getLastDisconnectedTimestamp()), and the proxy's own credential refresh lives in
ProxyConnection — the task #25179 refactored — not in this class.

So the branch is documented as PIP-478 routing-rule conformance and the description now says plainly
that nothing covers it and nothing can, instead of claiming coverage it never had.

2. Shared monitor on the connect path — taken

getProxyClientAuthenticationDriver() now reads the memoized driver without the monitor and
synchronizes only on the miss, matching ClientCnx.resolveAuthDriver. The field is volatile, as you
noted it has to be.

3. The command-build failure path also closes — said in the description

Added, including which checkState it is and what it used to do instead.

4. Auth method name — comment added

Stating why reading it from the plugin is right here and does not depend on a completed round, so the
next reader does not "fix" it into ClientCnx's shape.

5. Two silent paths — both addressed

whenCompleteAsync's discarded future now says it is discarded on purpose and why the only way it can
fail is already-dying. The !isActive() early return logs at debug, so "backend connected but never
sent CommandConnect" is diagnosable.

Round-state guarding — taken, minus one piece, deliberately

Your table is right that all three are missing. Two are now there:

• serialize-or-drop. A challenge arriving while a round is in flight is dropped. Your two-frames-
  in-one-read case is exactly the reachable one: during Init the handler decodes every frame, so
  both invocations land in the same event-loop turn before either resolution completes.
• the round cap. MAX_AUTH_CHALLENGE_ROUNDS, mirroring ClientCnx, closing the backend
  connection. Agreed this was the one to act on first — it is not a concurrency question, and the
  proxy is the more exposed component.

The generation guard I did not take, and I want to be explicit about the divergence. Once a
challenge arriving mid-round is dropped, rounds are strictly serialized: nothing left can supersede an
in-flight round, so there is no superseded continuation for a generation to detect. ClientCnx needs
it precisely because it does not drop everything — it lets a REFRESH supersede an in-flight round.
The proxy can drop a REFRESH instead, for two independent reasons: it cannot reach this handler at all
(§1), and the broker's refresh check is a scheduleAtFixedRate task, so a dropped one is re-sent on
the next tick. Adding a generation counter here would be state that no reachable path can exercise.
The comment in handleAuthChallenge records the divergence and the reason, so it reads as a decision
rather than as the third missing row of your table.

The test — went further than the recording plugin

Your suggestion was the right instinct and I took the diagnosis behind it: the fixture was the problem,
not the assertion. But an end-to-end proxy fixture cannot observe the round guards at all, and it makes
even the thread assertion timing-dependent. DirectProxyHandlerAuthTest builds ProxyBackendHandler
on an EmbeddedChannel instead — whose event loop runs a submitted task only when the test asks it to,
which turns "on the loop or off it" from a race into an assertion, and lets a challenge be delivered
while a resolution is deliberately still pending.

Four tests, each mutation-verified to fail alone and for its own reason:

Test	Mutation	Failure
theConnectCredentialIsResolvedOffTheEventLoop	inline the v4 call in channelActive	recorded thread is TestNG-method=…, not pulsar-auth-blocking-shared
theChallengeRoundIsResolvedOffTheEventLoop	inline the v4 call in handleAuthChallenge	both recorded threads are the event loop
aChallengeArrivingWhileARoundIsInFlightIsDropped	remove the drop guard	Expecting AtomicInteger(1) to have value: 0
anEndlesslyChallengingBrokerIsCutOffAtTheRoundCap	remove the cap	Expecting AtomicInteger(11) to have value: 10

The one property I did not write an assertion for is "command construction resumes on the event loop".
It is guaranteed by whenCompleteAsync(…, ctx.executor()), and I could not find an assertion for it
that would fail if the hop were removed — Netty accepts a write from any thread and orders it itself,
so the observable behaviour is identical either way. Rather than add a test that passes for both, I
left it argued from the code. Happy to be shown one.

On the shared pool, from your #26327 review

Recording where that lands now: the proxy still borrows the shared pool here, by design — it has no
client to lend one. What changed is the other end of the collision you described. With the admin off
the shared pool (#26327), a burst of admin requests against a stalled IdP can no longer occupy the
eight shared workers that this proxy's backend connection setup queues behind. Whether the proxy should
own a pool rather than borrow one is still open, and this is the PR to ask it in if you think it should
be.

quickCheck, sanityCheck and the proxy authentication suites pass.

_{Prepared with the assistance of Claude Code (Opus 5).}

[lhotari]

/pulsarbot rerun

[david-streamlio]

Verified d0a4cfece4 against everything outstanding. All of it is addressed, and the one place you diverged from what was asked is the place I'd have argued for.

The round-state guarding. authRoundInProgress and MAX_AUTH_CHALLENGE_ROUNDS are both in, and the flag's lifecycle is right where it would be easy to get wrong: set synchronously inside sendWhenResolved before the continuation is registered, cleared as the first statement of the continuation, which runs on ctx.executor() — so it is only ever touched on the event loop, like everything else in that class.

The generation guard you left out — I think that is correct, and it is a better answer than the one requested. Once a challenge arriving mid-round is dropped, rounds are strictly serialized, so there is no superseded continuation for a generation counter to detect. ClientCnx needs one only because it deliberately lets a REFRESH supersede an in-flight round; the proxy can drop that instead, and your two reasons for why it may are independent, which is what makes it safe rather than merely convenient — the sentinel cannot reach the handler at all, and the broker's refresh check being a scheduleAtFixedRate task means a dropped one is re-sent on the next tick. Adding the counter would have been state no reachable path could exercise. The comment recording the divergence is what makes this maintainable; without it the next reader sees two of three rows from ClientCnx and "fixes" the third.

Worth noting for anyone reading the thread later: this is a case where the bot's ask and the right fix diverged. Applying all three guards verbatim would have added dead state; the analysis of why only two apply is what produced the smaller, correct change.

The rest, all confirmed: the memoized driver is read without the monitor and the field is volatile; the command-build failure path and its checkState are described in the body; the auth-method-name comment is there; the discarded future says it is discarded on purpose; and the !isActive() return logs at debug so "backend connected but never sent CommandConnect" is diagnosable.

On the REFRESH branch — agreed with the sharper version. I had it as a timing argument and you are right that it is structural: after the handshake the connection stops decoding and switches to byte-level proxying, so the frame cannot reach handleAuthChallenge regardless of when the broker's task fires. Describing the branch as routing-rule conformance with nothing covering it, rather than claiming coverage, is the honest framing.

Nothing further from me. Approving separately.

[david-streamlio]

Approving. Verified all of it on d0a4cfece4; details in the comment above.

The round guard and cap are in, with the flag lifecycle correct — set synchronously in sendWhenResolved, cleared on the event-loop continuation. The generation guard was deliberately left out and I think that is the right call rather than a gap: dropping a mid-round challenge makes rounds strictly serialized, so there is nothing left to supersede, and the divergence is documented so the next reader does not restore it.

The five findings from my original review are all addressed, including the two silent paths.