You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/08/13 00:54:24 UTC
(pulsar) 03/03: [fix][client] PIP-478: review fixes for the Athenz and SASL v5 plugins
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-pip-478-auth-plugins-v3
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ae494527903239780b22246c01facce3fc988c76
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Aug 13 02:24:08 2026 +0300
[fix][client] PIP-478: review fixes for the Athenz and SASL v5 plugins
Three findings from reviewing this change, one of which was fixed in the base commit
because it was not specific to these plugins.
The SASL shim cached its HTTP authentication driver with whatever framework services were
bound at first use, and kept it until close. One plugin instance is routinely shared
between a PulsarClient and a PulsarAdmin, and both bind services — so whichever bound
first won, and the other transport ran with services meant for its neighbour. Rebuild the
driver when the binding changes; it holds no cross-request state, so replacing it is safe.
The rebase left two javadoc blocks stacked before toHeaders(), so the transport class and
completeAndClose() were both undocumented while toHeaders carried a doc describing neither
(including a @param for an argument it does not take). Each is back on the member it
describes. The Athenz shim similarly kept a comment about framework services it no longer
holds, and an editing artifact in its class javadoc.
completeAndClose's javadoc claimed it was package-private so the close contract could be
asserted on both branches — and nothing asserted it, so the response-leak fix this change
carries could have regressed silently. JaxRsResponseCloseTest now pins both branches:
completing from a response closes it, and a response arriving after the future already
settled is closed too. Mutation-verified — restoring the original close-only-on-the-late-
branch behaviour fails the first case.
Assisted-by: Claude Code (Opus 5)
---
.../client/impl/auth/AuthenticationAthenz.java | 7 +-
.../client/impl/auth/AuthenticationAthenzTest.java | 16 ++---
.../client/impl/auth/AuthenticationSasl.java | 72 +++++++++++--------
.../client/impl/auth/JaxRsResponseCloseTest.java | 83 ++++++++++++++++++++++
4 files changed, 135 insertions(+), 43 deletions(-)
diff --git a/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
index aba6c347157c..5842a481c0cd 100644
--- a/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
+++ b/pulsar-client-auth-athenz/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenz.java
@@ -62,9 +62,9 @@ import org.apache.pulsar.client.impl.auth.v5.AthenzAuthenticationV5;
*
* <p>The verbatim v4 synchronous surface ({@link #getAuthData()} / {@link AuthenticationDataAthenz}) is
* preserved for callers of the v4 API; PIP-478 additionally exposes the v5-native body through
- * {@link V5AuthenticationProvider}, which is what the client drives. The old note: it can drive
- * the role-token credential over the non-blocking binary path via the v5-native
- * {@link AthenzAuthenticationV5}. The ZTS role-token cache stays on this shim.
+ * {@link V5AuthenticationProvider} — the v5-native {@link AthenzAuthenticationV5}, which the client drives
+ * over the non-blocking binary path without bridging this class. The ZTS role-token exchange and its cache
+ * stay here, on the Athenz SDK's own transport.
*/
public class AuthenticationAthenz
implements Authentication, EncodedAuthenticationParameterSupport, V5AuthenticationProvider {
@@ -73,7 +73,6 @@ public class AuthenticationAthenz
private static final String APPLICATION_X_PEM_FILE = "application/x-pem-file";
- // PIP-478: the client's framework services, late-bound before start(); null until then.
private transient KeyRefresher keyRefresher = null;
private transient ZTSClient ztsClient = null;
diff --git a/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java b/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
index 91abafd599d5..acc63bb16e4b 100644
--- a/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
+++ b/pulsar-client-auth-athenz/src/test/java/org/apache/pulsar/client/impl/auth/AuthenticationAthenzTest.java
@@ -46,7 +46,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
@@ -373,16 +374,15 @@ public class AuthenticationAthenzTest {
final AuthenticationExchange exchange =
new V5BinaryAuthenticationDriver(V5AuthenticationLoader.forStartedV4Plugin(auth))
.newAuthenticationExchange("broker.example.com");
- // No blocking executor is bound in this test, so supplyBlocking runs the credential fetch inline
- // and the failure is produced deterministically here.
+ // The fetch is off-loaded even with no client services bound (credential acquisition never runs
+ // on the caller thread), so await the failure rather than expecting it to have happened already.
final CompletableFuture<AuthData> future = exchange.getAuthDataAsync();
- assertTrue(future.isCompletedExceptionally());
try {
- future.join();
+ future.get(10, TimeUnit.SECONDS);
fail("expected the ZTS failure to propagate");
- } catch (CompletionException ce) {
- assertTrue(ce.getCause() instanceof PulsarClientException.GettingAuthenticationDataException,
- "expected a v4 GettingAuthenticationDataException subtype, got: " + ce.getCause());
+ } catch (ExecutionException ee) {
+ assertTrue(ee.getCause() instanceof PulsarClientException.GettingAuthenticationDataException,
+ "expected a v4 GettingAuthenticationDataException subtype, got: " + ee.getCause());
}
auth.close();
diff --git a/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
index 47e6096d2f0d..3b139fc8fe42 100644
--- a/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
+++ b/pulsar-client-auth-sasl/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationSasl.java
@@ -102,6 +102,8 @@ public class AuthenticationSasl
// first HTTP request after start(). Its default transport is this plugin's own JAX-RS client (the
// faithful admin-path transport; the lookup path supplies its shared AsyncHttpClient instead).
private transient volatile HttpAuthenticationDriver httpAuthenticationDriver;
+ // The services the cached driver was built with, so a later rebind is noticed rather than ignored.
+ private transient volatile ClientAuthenticationServices httpAuthenticationDriverServices;
public AuthenticationSasl() {
}
@@ -144,31 +146,40 @@ public class AuthenticationSasl
// HTTP callers route the 401->resubmit->200 exchange through the shared state machine instead of the
// deprecated authenticationStage(...) hook (which stays for third-party v4 plugins). The default
// transport is this plugin's own JAX-RS client — exactly what authenticationStage(...) uses today.
+ // The driver captures the services it is built with, so caching it across a rebind would freeze
+ // whichever binding happened to come first. One plugin instance is routinely shared between a
+ // PulsarClient and a PulsarAdmin, and both bind: caching the first would hand one of them the
+ // other's services for the rest of its life. Rebuild when the binding changes — the driver holds no
+ // cross-request state (that lives in the per-request call context), so replacing it is safe.
+ ClientAuthenticationServices services = this.authServices;
HttpAuthenticationDriver driver = httpAuthenticationDriver;
- if (driver == null) {
- synchronized (this) {
- driver = httpAuthenticationDriver;
- if (driver == null) {
- driver = new HttpAuthenticationDriver(
- new SaslAuthenticationV5(new ShimSaslProviderFactory(this)),
- authServices, new JaxRsChallengeTransport());
- httpAuthenticationDriver = driver;
- }
+ if (driver != null && httpAuthenticationDriverServices == services) {
+ return Optional.of(driver);
+ }
+ synchronized (this) {
+ driver = httpAuthenticationDriver;
+ if (driver == null || httpAuthenticationDriverServices != services) {
+ driver = new HttpAuthenticationDriver(
+ new SaslAuthenticationV5(new ShimSaslProviderFactory(this)),
+ services, new JaxRsChallengeTransport());
+ httpAuthenticationDriver = driver;
+ httpAuthenticationDriverServices = services;
}
+ return Optional.of(driver);
}
- return Optional.of(driver);
}
- /**
- * A {@link HttpChallengeTransport} backed by this plugin's own JAX-RS {@link Client} (created in
- * {@link #start()}) — the faithful transport for the admin-path SASL warmup, matching what
- * {@code authenticationStage(...)} does today. Each round is a bodiless {@code GET} to the original URI.
- * The JAX-RS client itself is created with default, unbounded timeouts, so the per-request {@code timeout}
- * is enforced here by bounding the returned future ({@link CompletableFuture#orTimeout}) AND cancelling the
- * underlying JAX-RS {@link Future} on timeout/failure — the request is bounded, so a peer that accepts the
- * connection but never responds cannot leak an in-flight request (fd/socket exhaustion) across retries. A
- * response that arrives after the future already timed out is closed rather than leaked.
- */
+ /** Copy a response's first-value headers into the v5 header carrier. */
+ private static HttpAuthHeaders toHeaders(Response response) {
+ Map<String, String> headers = new LinkedHashMap<>();
+ response.getStringHeaders().forEach((name, values) -> {
+ if (values != null && !values.isEmpty() && values.get(0) != null) {
+ headers.put(name, values.get(0));
+ }
+ });
+ return HttpAuthHeaders.of(headers);
+ }
+
/**
* Complete {@code future} from {@code response} and always close the response.
*
@@ -184,17 +195,6 @@ public class AuthenticationSasl
* @param future the future to complete
* @param response the JAX-RS response, always closed before returning
*/
-
- private static HttpAuthHeaders toHeaders(Response response) {
- Map<String, String> headers = new LinkedHashMap<>();
- response.getStringHeaders().forEach((name, values) -> {
- if (values != null && !values.isEmpty() && values.get(0) != null) {
- headers.put(name, values.get(0));
- }
- });
- return HttpAuthHeaders.of(headers);
- }
-
static void completeAndClose(CompletableFuture<HttpChallengeTransport.Result> future, Response response) {
try {
future.complete(new HttpChallengeTransport.Result(response.getStatus(), toHeaders(response)));
@@ -203,6 +203,16 @@ public class AuthenticationSasl
}
}
+ /**
+ * A {@link HttpChallengeTransport} backed by this plugin's own JAX-RS {@link Client} (created in
+ * {@link #start()}) — the faithful transport for the admin-path SASL warmup, matching what
+ * {@code authenticationStage(...)} does today. Each round is a bodiless {@code GET} to the original URI.
+ * The JAX-RS client itself is created with default, unbounded timeouts, so the per-request {@code timeout}
+ * is enforced here by bounding the returned future ({@link CompletableFuture#orTimeout}) AND cancelling the
+ * underlying JAX-RS {@link Future} on timeout/failure — the request is bounded, so a peer that accepts the
+ * connection but never responds cannot leak an in-flight request (fd/socket exhaustion) across retries. A
+ * response that arrives after the future already timed out is closed rather than leaked.
+ */
private final class JaxRsChallengeTransport implements HttpChallengeTransport {
@Override
public CompletableFuture<Result> get(URI uri, HttpAuthHeaders requestHeaders, Duration timeout) {
diff --git a/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/JaxRsResponseCloseTest.java b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/JaxRsResponseCloseTest.java
new file mode 100644
index 000000000000..0ac7ceda2109
--- /dev/null
+++ b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/JaxRsResponseCloseTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import jakarta.ws.rs.core.MultivaluedHashMap;
+import jakarta.ws.rs.core.MultivaluedMap;
+import jakarta.ws.rs.core.Response;
+import java.util.concurrent.CompletableFuture;
+import org.apache.pulsar.client.impl.auth.v5.HttpChallengeTransport;
+import org.testng.annotations.Test;
+
+/**
+ * PIP-478: the SASL-over-HTTP transport must close every JAX-RS {@link Response} it completes from.
+ *
+ * <p>{@code InvocationCallback<Response>} hands the caller an unclosed response, and reading only its
+ * headers neither consumes the entity nor releases the connection. The authentication driver runs at least
+ * one round on every admin request, so a response left open leaks a pooled connection per request rather
+ * than per client — which is why the completion path, not only the timed-out path, has to close.
+ */
+public class JaxRsResponseCloseTest {
+
+ @Test
+ public void completingFromAResponseAlsoClosesIt() {
+ Response response = responseWith(401, "WWW-Authenticate", "Negotiate");
+ CompletableFuture<HttpChallengeTransport.Result> future = new CompletableFuture<>();
+
+ AuthenticationSasl.completeAndClose(future, response);
+
+ assertThat(future).isCompleted();
+ HttpChallengeTransport.Result result = future.join();
+ assertThat(result.statusCode()).isEqualTo(401);
+ // HttpAuthHeaders normalises names, since HTTP header names are case-insensitive.
+ assertThat(result.responseHeaders().asMap()).containsEntry("www-authenticate", "Negotiate");
+ verify(response, times(1)).close();
+ }
+
+ /**
+ * The response has to be released even when the future has already been completed elsewhere — the round
+ * timed out, or the request was cancelled and the peer answered afterwards. Completing a settled future
+ * is a no-op, so without the close in a finally the late response would leak silently.
+ */
+ @Test
+ public void aResponseArrivingAfterTheFutureSettledIsStillClosed() {
+ Response response = responseWith(200, "X-Ignored", "value");
+ CompletableFuture<HttpChallengeTransport.Result> future = new CompletableFuture<>();
+ future.completeExceptionally(new IllegalStateException("round already timed out"));
+
+ AuthenticationSasl.completeAndClose(future, response);
+
+ assertThat(future).isCompletedExceptionally();
+ verify(response, times(1)).close();
+ }
+
+ private static Response responseWith(int status, String headerName, String headerValue) {
+ Response response = mock(Response.class);
+ MultivaluedMap<String, String> headers = new MultivaluedHashMap<>();
+ headers.putSingle(headerName, headerValue);
+ when(response.getStatus()).thenReturn(status);
+ when(response.getStringHeaders()).thenReturn(headers);
+ return response;
+ }
+}