[improve][client] PIP-478: Athenz and SASL v5 authentication plugins

[lhotari]

Main Issue: #25890

PIP: #25890

Stacked on #26317 — this PR's base is lh-pip-478-v5-native-auth-v2, the branch of the v5-native auth inversion. Review that one first; the diff here shows only this part.

Motivation

PIP-478 migrates Pulsar's built-in authentication plugins onto an asynchronous, capability-segregated v5 SPI. Token, basic and OAuth2 landed with the earlier PRs; Athenz and SASL are the two that remain, and they are the ones that made the design earn its keep.

SASL is the interesting case. It is multi-round on both transports, and SaslAuthenticationV5 is the first production implementor of the framework HTTP authentication driver the core migration added — the piece that makes HttpAuthenticationDriver / AsyncHttpAuthenticationProvider a live extension point rather than an unused one. It is also the plugin whose credential work is most worth getting off the event loop: a GSSAPI exchange talks to a KDC.

Modifications

AthenzAuthenticationV5 — a single-pass role-token credential over both transports. The ZTS exchange and its cache stay on the v4 shim, which owns the Athenz SDK's transport; the body reads the current role token through a provider. This is the layering PIP-478 specifies for the credential-acquisition-heavy plugins: expose the async surface without reimplementing hard-won provider logic.

SaslAuthenticationV5 — BinaryAuthDataProvider + BinaryAuthChallengeHandler for the binary protocol, HttpAuthChallengeHandler + HttpAuthHeadersProvider for SASL over HTTP. The per-broker PulsarSaslClient lives in the exchange's call-context state slot, so one body serves the whole client while each connection keeps its own handshake state, and concurrent handshakes to different brokers cannot collide.

Both shims hand their body over through V5AuthenticationProvider, the seam #26317 introduced, so every built-in now works the same way and the seam's javadoc no longer has to except two of them.

Two defects fixed alongside, both found reviewing the original version of this change:

• JaxRsChallengeTransport leaked every successfully-completed JAX-RS Response. InvocationCallback<Response> hands the caller an unclosed response, and reading only its headers neither consumes the entity nor releases the connection — so the success branch leaked one pooled connection per authentication round while only the timed-out branch closed. The driver runs at least one round on every admin request, so this was per request, not per client.
• AuthenticationSasl.client and saslRoleToken were plain fields, written by start()/close() on the application thread and read from the challenge driver's Jersey continuation threads with no happens-before edge.

And three from reviewing this rebase:

• The shim cached its HTTP authentication driver with whatever framework services were bound at first use. One plugin instance is routinely shared between a PulsarClient and a PulsarAdmin, and both bind — so whichever bound first won, and the other transport ran with services meant for its neighbour. The driver is rebuilt when the binding changes; it holds no cross-request state.
• completeAndClose's javadoc claimed it was package-private so the close contract could be asserted, and nothing asserted it — the leak fix above could have regressed silently. JaxRsResponseCloseTest now pins both branches, mutation-verified against the original behaviour.
• Two javadoc blocks had ended up stacked on the wrong members, leaving the transport class and completeAndClose undocumented while toHeaders carried a description of neither.

Verifying this change

This change added tests and can be verified as follows:

• SaslAuthenticationV5BinaryOffloadTest — the per-exchange SASL provider creation and evaluateChallenge run on the blocking executor, not the caller thread. Driven through the real V5BinaryAuthenticationDriver with a deliberately-blocking fake provider.
• SaslAuthenticationV5HttpTest — the SASL-over-HTTP 401 → resubmit → 200 exchange through the framework driver.
• JaxRsResponseCloseTest — every completion path closes its response, including one arriving after the future already settled. Mutation-verified.
• AuthenticationAthenzTest — the async path preserves the GettingAuthenticationDataException subtype, driven through the production resolution path rather than a test-only seam.

:pulsar-client-auth-athenz:test, :pulsar-client-auth-sasl:test, :pulsar-client-v5:test, :pulsar-client-original:test, quickCheck and sanityCheck pass locally; full CI green on the equivalent branch.

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

The schema box is checked only to be safe: no serialized format changes, but AuthenticationSasl and AuthenticationAthenz are Serializable public classes whose fields changed (added volatile, added the driver cache). Their serialVersionUID is unchanged and no field was removed or retyped.

The threading model: Athenz and SASL credential work now runs on a blocking executor rather than on the calling thread. On the client that thread was already an executor; on paths with no client-owned executor — the proxy's broker connections — it now uses the shared fallback pool introduced in #26317 rather than the caller's Netty event loop.

Documentation

• doc-required
• doc-not-needed
• doc
• doc-complete

Internal migration of two built-in plugins; no configuration or user-facing API changes.

Matching PR in forked repository

PR in forked repository: lhotari#253

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

[Copilot]

Pull request overview

Migrates Athenz and SASL authentication to the asynchronous v5 SPI while preserving v4 compatibility.

Changes:

• Adds v5-native Athenz and multi-round SASL implementations.
• Offloads blocking credential work and adds HTTP challenge handling.
• Adds coverage for HTTP exchanges, executor offloading, exception preservation, and response cleanup.

Reviewed changes

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

Show a summary per file

File	Description
pulsar-client-auth-sasl/.../SaslAuthenticationV5HttpTest.java	Tests SASL-over-HTTP exchanges and token caching.
pulsar-client-auth-sasl/.../SaslAuthenticationV5BinaryOffloadTest.java	Tests binary SASL executor offloading.
pulsar-client-auth-sasl/.../JaxRsResponseCloseTest.java	Tests JAX-RS response cleanup.
pulsar-client-auth-sasl/.../SaslAuthenticationV5.java	Implements v5 SASL authentication.
pulsar-client-auth-sasl/.../v5/package-info.java	Documents the SASL v5 package.
pulsar-client-auth-sasl/.../AuthenticationSasl.java	Bridges v4 SASL to v5 drivers.
pulsar-client-auth-sasl/build.gradle.kts	Adds framework test dependency.
pulsar-client-auth-athenz/.../AuthenticationAthenzTest.java	Tests asynchronous exception preservation.
pulsar-client-auth-athenz/.../v5/package-info.java	Documents the Athenz v5 package.
pulsar-client-auth-athenz/.../AthenzAuthenticationV5.java	Implements v5 Athenz authentication.
pulsar-client-auth-athenz/.../AuthenticationAthenz.java	Exposes the Athenz v5 body.
pulsar-client-api-v5/.../V5AuthenticationProvider.java	Updates provider documentation.

---

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

[david-streamlio]

Reviewed against master (this retargeted cleanly once #26317 merged, so it's a self-contained 12-file diff). The Athenz layering is clean, and the SASL body is a careful port. Three things below, two of which are about the two new fixes interacting.

A few things I checked and can confirm rather than just take on trust:

• The CompletionException reasoning in AuthenticationAthenz.currentRoleToken() is exactly right. CompletableFuture's AsyncSupply calls encodeThrowable, which re-uses an already-CompletionException throwable rather than wrapping it again — so exactly one layer reaches BinaryAuthenticationExchange.unwrap, and GettingAuthenticationDataException survives to toV4Exception. A bare RuntimeException really would have flattened it. The comment explaining this is accurate, which is worth saying because it is the kind of claim that is usually slightly wrong.
• The JAX-RS leak fix is sound. orTimeout mutates and returns this, so discarding the return value still arms the timeout on the returned future; completeAndClose's finally covers the late-arrival branch that complete() no-ops. Good catch on the original — one pooled connection per admin request is a real leak.
• The HTTP port quietly fixes a latent v4 NPE. v4 does previousRespHeaders.get(SASL_HEADER_STATE).equalsIgnoreCase(SASL_STATE_COMPLETE), which NPEs when the server omits that header; the port writes it constant-first. Unclaimed in the description, worth keeping.

1. httpAuthenticationDriver()'s fast path can still hand back a driver built with the other binding's services.

The fast path reads two volatiles independently:

ClientAuthenticationServices services = this.authServices;
HttpAuthenticationDriver driver = httpAuthenticationDriver;      // read A
if (driver != null && httpAuthenticationDriverServices == services) {   // read B

and the writer publishes them in sequence under the lock:

httpAuthenticationDriver = driver;              // write 1
httpAuthenticationDriverServices = services;    // write 2

A reader that performs read A before write 1 and read B after write 2 sees the old driver paired with the new services, passes the guard, and returns the driver built with the previous binding — precisely the defect this fix exists to remove. It self-heals on the next call, so the blast radius is one request. But since both call sites (BaseResource:131, HttpClient:358) invoke this per request, "two threads at once" is the normal case for a shared client+admin.

The interleaving:

#	reader thread	rebuilding thread
1	services = authServices → S2
2	driver → D1 (read A)
3		httpAuthenticationDriver = D2 (write 1)
4		httpAuthenticationDriverServices = S2 (write 2)
5	driverServices == services → S2 == S2 ✓ (read B)
6	returns D1, built with S1

Note this is not a memory-model subtlety — volatile accesses are totally ordered, so no reordering is involved. It is a plain temporal window: read A simply happens before write 1, and read B after write 2.

Reproduction. The window is sub-microsecond, so I built a harness that mirrors the exact field-access shape and runs it two ways — pinned (deterministic) and unassisted. Saved as a single file, runs with java BindingRace.java, no dependencies:

BindingRace.java

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

/** Repro for the httpAuthenticationDriver() binding race. Run: java BindingRace.java */
public class BindingRace {
    record Services(String name) { public String toString() { return name; } }
    record Driver(Services builtWith) { }
    record Result(Driver driver, Services comparedAgainst) { }
    static volatile boolean stop;

    /** The shape as written in AuthenticationSasl: two independently-read volatiles. */
    static class Racy {
        volatile Services authServices; volatile Driver driver; volatile Services driverServices;
        volatile Thread pin; volatile Runnable betweenReads;
        Result get() {
            Services services = authServices;
            Driver d = driver;                                     // read A
            if (Thread.currentThread() == pin && betweenReads != null) betweenReads.run();
            if (d != null && driverServices == services) return new Result(d, services);  // read B
            synchronized (this) {
                d = driver;
                if (d == null || driverServices != services) {
                    Driver nd = new Driver(services);
                    driver = nd;                                   // write 1
                    driverServices = services;                     // write 2
                    d = nd;
                }
                return new Result(d, services);
            }
        }
    }

    /** The fix: one immutable pair behind one volatile, so the fast path is a single read. */
    static class Fixed {
        record Binding(Driver driver, Services services) { }
        volatile Services authServices; volatile Binding binding;
        volatile Thread pin; volatile Runnable betweenReads;
        Result get() {
            Services services = authServices;
            Binding b = binding;                                   // single read
            if (Thread.currentThread() == pin && betweenReads != null) betweenReads.run();
            if (b != null && b.services() == services) return new Result(b.driver(), services);
            synchronized (this) {
                b = binding;
                if (b == null || b.services() != services) binding = b = new Binding(new Driver(services), services);
                return new Result(b.driver(), services);
            }
        }
    }

    /** Pin a reader between its two reads while another thread rebuilds. Deterministic. */
    static Result pinned(Object holder) throws Exception {
        Services s1 = new Services("S1-client"), s2 = new Services("S2-admin");
        Racy r = holder instanceof Racy x ? x : null; Fixed f = holder instanceof Fixed x ? x : null;
        if (r != null) { r.authServices = s1; r.get(); r.authServices = s2; }
        else { f.authServices = s1; f.get(); f.authServices = s2; }
        CountDownLatch didReadA = new CountDownLatch(1), writerDone = new CountDownLatch(1);
        Result[] out = new Result[1];
        Thread reader = new Thread(() -> out[0] = r != null ? r.get() : f.get());
        Runnable hook = () -> { didReadA.countDown(); try { writerDone.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } };
        if (r != null) { r.pin = reader; r.betweenReads = hook; } else { f.pin = reader; f.betweenReads = hook; }
        reader.start();
        didReadA.await();                                          // reader has done read A
        new Thread(() -> { if (r != null) r.get(); else f.get(); writerDone.countDown(); }).start();
        reader.join();
        return out[0];
    }

    /** No hooks at all: concurrent readers plus a thread that rebinds. */
    static long stress(Object holder, int threads, long ms) throws Exception {
        Services s1 = new Services("S1-client"), s2 = new Services("S2-admin");
        Racy r = holder instanceof Racy x ? x : null; Fixed f = holder instanceof Fixed x ? x : null;
        if (r != null) { r.authServices = s1; r.get(); } else { f.authServices = s1; f.get(); }
        AtomicLong stale = new AtomicLong(), calls = new AtomicLong();
        stop = false;
        Thread flip = new Thread(() -> { boolean a = false; while (!stop) { Services s = (a = !a) ? s2 : s1; if (r != null) r.authServices = s; else f.authServices = s; Thread.onSpinWait(); } });
        flip.setDaemon(true); flip.start();
        Thread[] ts = new Thread[threads];
        for (int i = 0; i < threads; i++) {
            ts[i] = new Thread(() -> { long n = 0; while (!stop) { Result res = r != null ? r.get() : f.get(); if (res.driver().builtWith() != res.comparedAgainst()) stale.incrementAndGet(); n++; } calls.addAndGet(n); });
            ts[i].setDaemon(true); ts[i].start();
        }
        Thread.sleep(ms); stop = true;
        for (Thread t : ts) t.join(2000);
        System.out.printf("    %,d calls, %,d stale pairings%n", calls.get(), stale.get());
        return stale.get();
    }

    static void show(String label, Result res) {
        System.out.printf("    %-6s compared against %-9s got driver built with %-9s -> %s%n", label,
                res.comparedAgainst(), res.driver().builtWith(),
                res.driver().builtWith() != res.comparedAgainst() ? "STALE (bug)" : "ok");
    }

    public static void main(String[] args) throws Exception {
        int n = Math.max(4, Runtime.getRuntime().availableProcessors());
        System.out.println("Deterministic (reader pinned between read A and read B):");
        show("racy", pinned(new Racy()));
        show("fixed", pinned(new Fixed()));
        System.out.println("Unassisted stress (" + n + " readers, 3s, no hooks):");
        System.out.println("  racy:");  stress(new Racy(), n, 3000);
        System.out.println("  fixed:"); stress(new Fixed(), n, 3000);
    }
}

Results on JDK 24 / arm64 (10 readers, 3s per mode):

Deterministic (reader pinned between read A and read B):
    racy   compared against S2-admin  got driver built with S1-client -> STALE (bug)
    fixed  compared against S2-admin  got driver built with S2-admin  -> ok
Unassisted stress (10 readers, 3s, no hooks):
  racy:   96,079,721 calls, 389 stale pairings
  fixed:  97,837,193 calls, 0 stale pairings

Across four runs the racy shape produced 377–746 stale pairings per ~100M calls (~1 in 200k); the fixed shape produced 0 across ~350M calls. The invariant checked is the precise one — the returned driver must have been built with the same services value the method compared against — so there are no false positives from authServices merely changing concurrently.

Two honest caveats: the stress mode rebinds continuously, which inflates the rate well above production, where rebinds cluster around client/admin construction. What it demonstrates is reachability without any injected hooks; the deterministic mode is what pins the interleaving itself. And this is the extracted shape, not AuthenticationSasl — asserting the invariant against the real class needs a way to see which services a driver was built with, i.e. a @VisibleForTesting accessor on HttpAuthenticationDriver. Probably not worth adding if you take the fix, since it makes the state unrepresentable.

Fix. Collapsing the pair into one immutable value behind a single volatile — a record Binding(HttpAuthenticationDriver driver, ClientAuthenticationServices services) — makes the fast path a single read and removes the interleaving by construction. That is the Fixed variant above, and it is what reports 0.

2. The rebuild-on-rebind fix and FIX C contradict each other.

The rebuild is justified as: "the driver holds no cross-request state (that lives in the per-request call context), so replacing it is safe."

That is no longer true in this PR. HttpAuthenticationDriver holds private final Authentication v5, and the body it holds is a fresh SaslAuthenticationV5 whose cachedRoleToken is — per its own comment — "the cross-request SASL-over-HTTP role-token cache". So rebuilding the driver discards a validated role token and forces the next request into a full Kerberos negotiation, which is the exact cost FIX C was added to avoid.

The impact is bounded (a rebind happens at client/admin construction, not per request), so this is a coherence problem more than a hot-path one. But the comment will be read as licence to rebuild freely, and it no longer holds.

There's a related consequence worth deciding on deliberately: the shim still carries saslRoleToken for the v4 newRequestHeader/getHeaders path, so a plugin instance now has two independent role-token caches that never share. A deployment exercising both paths negotiates Kerberos twice.

Both fall out if the cache lives on the shim rather than on the body — which is where v4 kept it, and saslRoleToken is already there and already volatile after this PR. The body would read/write it through the same SaslProviderFactory-style seam it already uses for the provider. That restores the comment's truth, survives rebuilds, and collapses the two caches into one.

3. The HTTP port drops v4's hasDataForHttp() guard.

v4:

if (authData.hasDataForHttp()) {
    authData.getHttpHeaders().forEach(...);
}

port:

conv.provider.getHttpHeaders().forEach(e -> headers.put(e.getKey(), e.getValue()));

AuthenticationDataProvider.getHttpHeaders() defaults to returning null (and hasDataForHttp() to false), so this NPEs for any provider that doesn't override both. Harmless for the built-in path — SaslAuthenticationDataProvider returns true and a non-null set — but SaslAuthenticationV5's constructor and the SaslProviderFactory interface are both public, so the guard isn't purely defensive. Cheap to restore.

Minor / worth confirming: in JaxRsChallengeTransport.get, the whenComplete cancels with responseFuture.cancel(true) and is guarded on !responseFuture.isDone(). That guard assumes Jersey marks its Future done before invoking InvocationCallback.completed(...). If it doesn't, a successful round would issue an interrupting cancel against the worker thread that just delivered the response. cancel(false) would be immune to the ordering either way. I didn't chase Jersey's ordering, so flagging rather than asserting.

For disclosure: this is static analysis over the branch — I did not run the suites or sanityCheck locally.

[david-streamlio]

Amending my own advice on finding 1, having now read the Copilot review that landed just before mine.

Its comment on AuthenticationSasl.java (#discussion_r3778725175) is the more fundamental version of what I reported, and my suggested fix is necessary but not sufficient.

I raised a race in the fast path: a reader interleaving between the two volatile writes can pair the old driver with the new services. Collapsing them into a single record Binding(driver, services) closes that. But Copilot's point stands independently of any race — bindClientAuthenticationServices writes one plugin-wide slot (AuthenticationSasl:131), and both PulsarClientImpl:415 and PulsarAdminImpl:590 write it. After both owners bind, authServices permanently holds whichever bound last, so the client's HTTP lookup path (HttpClient:358) reads the admin's services and runs SASL work with the admin's executor and telemetry. The fix in this PR turned "first binding wins" into "last binding wins"; it did not make the driver per-owner.

So please don't read my record Binding suggestion as the whole remedy — on its own it would close the window I demonstrated while leaving the steady-state mismatch in place, which is the worse of the two. Keying the driver to the owner (captured when that owner binds, or the caller's services passed through explicitly) fixes both, and makes my interleaving unrepresentable rather than merely unlikely.

My other two findings are unaffected. Finding 2 (the rebuild-on-rebind comment vs. cachedRoleToken being cross-request state) actually gets more pointed under a per-owner design, since there would then be one body per owner and the placement of the role-token cache matters more, not less.

Copilot's other three comments — the stale binary-only wording in the SASL v5 package-info, and the "runs inline" claims on AthenzAuthenticationV5 and SaslAuthenticationV5 — all check out against the code. V5AuthContexts.supplyBlocking substitutes sharedBlockingExecutor() when the executor is null, so that work never degrades to inline; those two comments describe a threading behaviour that cannot occur. Worth fixing precisely because they are the comments a reader consults to answer "can this block the event loop?".