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 09:29:02 UTC
(pulsar) 01/05: [improve][client] PIP-478: move the binary auth driver out of the public API module
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-pip-478-followups
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 25803f3184b60f62917f112f95df50c0a2a58e65
Author: Lari Hotari <lh...@apache.org>
AuthorDate: Thu Aug 13 03:25:51 2026 +0300
[improve][client] PIP-478: move the binary auth driver out of the public API module
AsyncAuthenticationDriver lived in pulsar-client-api as a "stable internal" marker, because
it existed for a carve-out: a v4 Authentication could additionally implement it and
ClientCnx would then drive that plugin asynchronously instead of synchronously. Since the
client drives the v5 model natively there is no such choice — the client resolves the
authentication it drives and wraps it itself, and nothing outside pulsar-client implements
or observes the type.
Move it to org.apache.pulsar.client.impl.auth.v5 and name it for what it now is,
BinaryAuthenticationDriver. That removes an internal type from the API module's surface, and
its javadoc no longer has to describe a synchronous alternative that no longer exists.
Also make the client's own blocking auth executor queue rather than reject. It was a
SynchronousQueue with the default abort policy, so the seventeenth concurrent credential call
failed outright — and every caller is a connection attempt or an authenticated request, so a
reconnect storm against a slow identity provider produced failed connections rather than slow
ones. This is the same fix the shared fallback pool already received, and it is what
pip-478.md already claims the executor does.
Assisted-by: Claude Code (Opus 5)
---
.../pulsar/client/api/internal/package-info.java | 2 +-
.../client/impl/auth/AuthenticationAthenzTest.java | 2 +-
.../v5/SaslAuthenticationV5BinaryOffloadTest.java | 2 +-
.../v5/CredentialOffloadThroughV5BuilderTest.java | 2 +-
.../client/impl/v5/V5AuthenticationLoaderTest.java | 2 +-
.../org/apache/pulsar/client/impl/ClientCnx.java | 20 ++++++++++----------
.../apache/pulsar/client/impl/PulsarClientImpl.java | 12 +++++++++---
.../auth/v5/AsyncHttpAuthenticationProvider.java | 2 +-
.../impl/auth/v5/BinaryAuthenticationDriver.java | 20 +++++++++-----------
.../impl/auth/v5/BinaryAuthenticationExchange.java | 2 +-
.../impl/auth/v5/V5BinaryAuthenticationDriver.java | 5 ++---
.../pulsar/client/impl/auth/v5/package-info.java | 2 +-
.../client/impl/conf/ClientConfigurationData.java | 4 ++--
.../pulsar/client/impl/ClientCnxAsyncAuthTest.java | 16 ++++++++--------
.../impl/auth/v5/AuthInitializationRetryTest.java | 2 +-
.../auth/v5/BinaryAuthenticationExchangeTest.java | 2 +-
.../client/impl/auth/v5/CredentialOffloadTest.java | 2 +-
17 files changed, 51 insertions(+), 48 deletions(-)
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/package-info.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/package-info.java
index 3828c2ff7808..8ebcdd629dc7 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/package-info.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/package-info.java
@@ -19,6 +19,6 @@
/**
* Stable-internal client API types observed by the framework but not intended for application
- * implementation. See {@link org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver}.
+ * implementation. See {@link org.apache.pulsar.client.api.internal.BinaryAuthenticationDriver}.
*/
package org.apache.pulsar.client.api.internal;
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 acc63bb16e4b..6c2cf56e3cf2 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
@@ -50,7 +50,7 @@ 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;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader;
import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver;
import org.apache.pulsar.common.api.AuthData;
diff --git a/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5BinaryOffloadTest.java b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5BinaryOffloadTest.java
index 12d043a3e676..97a2ebc325e0 100644
--- a/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5BinaryOffloadTest.java
+++ b/pulsar-client-auth-sasl/src/test/java/org/apache/pulsar/client/impl/auth/v5/SaslAuthenticationV5BinaryOffloadTest.java
@@ -30,8 +30,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServices;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.common.api.AuthData;
import org.testng.annotations.Test;
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/CredentialOffloadThroughV5BuilderTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/CredentialOffloadThroughV5BuilderTest.java
index cf613b2ed8e4..8ce1cec18030 100644
--- a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/CredentialOffloadThroughV5BuilderTest.java
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/CredentialOffloadThroughV5BuilderTest.java
@@ -31,8 +31,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServices;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.auth.v5.DefaultClientAuthenticationServices;
import org.apache.pulsar.client.impl.auth.v5.LegacyV4AuthenticationAdapter;
import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver;
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5AuthenticationLoaderTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5AuthenticationLoaderTest.java
index 83953f88242c..47d341c19f42 100644
--- a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5AuthenticationLoaderTest.java
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5AuthenticationLoaderTest.java
@@ -23,7 +23,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.auth.Authentication;
import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext;
@@ -32,6 +31,7 @@ import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
import org.apache.pulsar.client.api.v5.auth.HttpAuthCallContext;
import org.apache.pulsar.client.api.v5.auth.HttpAuthHeaders;
import org.apache.pulsar.client.api.v5.auth.SinglePassAuthentication;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.auth.v5.LegacyV4AuthenticationAdapter;
import org.apache.pulsar.client.impl.auth.v5.NoAuthentication;
import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
index 6ca84ccf470c..9cb38dd4703e 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
@@ -61,9 +61,9 @@ import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.ConnectException;
import org.apache.pulsar.client.api.PulsarClientException.TimeoutException;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.BinaryProtoLookupService.LookupDataResult;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader;
import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -132,7 +132,7 @@ public class ClientCnx extends PulsarHandler {
// PIP-478: the authentication the client drives. The client resolves exactly one v5 Authentication and
// wraps it in this per-client driver, from which every connection attempt opens its own exchange; there
// is no synchronous plugin path left on the connect/challenge routes.
- protected final AsyncAuthenticationDriver authDriver;
+ protected final BinaryAuthenticationDriver authDriver;
protected State state;
@VisibleForTesting
@@ -255,7 +255,7 @@ public class ClientCnx extends PulsarHandler {
@Getter
protected AuthenticationDataProvider authenticationDataProvider;
// PIP-478: the authentication exchange for the current connection attempt. Non-null only while a
- // connect driven through an AsyncAuthenticationDriver is in progress; a fresh exchange is created per
+ // connect driven through an BinaryAuthenticationDriver is in progress; a fresh exchange is created per
// connect attempt and whenever the broker pushes the REFRESH sentinel.
private AuthenticationExchange authenticationExchange;
// PIP-478: guards for the async auth carve-out. Both are touched only on the channel's event loop
@@ -388,7 +388,7 @@ public class ClientCnx extends PulsarHandler {
}
// Send CONNECT command. Sync and async plugins share a single state machine (PIP-478): the initial
// credential is resolved as a CompletableFuture — already-completed for a plain v4 plugin (computed
- // inline, preserving today's behaviour verbatim) or pending for an AsyncAuthenticationDriver — and
+ // inline, preserving today's behaviour verbatim) or pending for an BinaryAuthenticationDriver — and
// fed to one continuation that assigns the data provider, builds the command, transitions state and
// writes. The continuation runs inline when the future is already done and hops to the channel's
// event executor only when it was pending.
@@ -401,7 +401,7 @@ public class ClientCnx extends PulsarHandler {
/**
* Resolve the initial credential for {@code CommandConnect}. For a plain v4 plugin this is computed
* inline (exactly as the previous synchronous connect path did) into an already-completed future so
- * the continuation runs synchronously on this thread; for an {@link AsyncAuthenticationDriver} it is a
+ * the continuation runs synchronously on this thread; for an {@link BinaryAuthenticationDriver} it is a
* fresh exchange's {@link AuthenticationExchange#getAuthDataAsync()}. The resolution carries the
* {@link AuthenticationDataProvider} to publish — the real v4 provider for the sync path (so subsequent
* synchronous challenge rounds keep working), or a minimal command-data adapter for the async path.
@@ -473,8 +473,8 @@ public class ClientCnx extends PulsarHandler {
* @param conf the client configuration this connection is built from
* @return the driver to open authentication exchanges against
*/
- private static AsyncAuthenticationDriver resolveAuthDriver(ClientConfigurationData conf) {
- AsyncAuthenticationDriver resolved = conf.getV5AuthenticationDriver();
+ private static BinaryAuthenticationDriver resolveAuthDriver(ClientConfigurationData conf) {
+ BinaryAuthenticationDriver resolved = conf.getV5AuthenticationDriver();
if (resolved != null) {
return resolved;
}
@@ -732,7 +732,7 @@ public class ClientCnx extends PulsarHandler {
* Drive an authentication continuation. To preserve the synchronous path verbatim (and never touch
* {@code ctx.executor()} for a plain v4 plugin), the continuation runs inline when the resolution future
* is already complete; it hops onto the channel's event executor only when the resolution was pending
- * (an {@link AsyncAuthenticationDriver} performing off-event-loop credential I/O).
+ * (an {@link BinaryAuthenticationDriver} performing off-event-loop credential I/O).
*/
private void driveAuthResolution(AuthRound round, CompletableFuture<ResolvedAuthData> resolution,
BiConsumer<ResolvedAuthData, Throwable> continuation) {
@@ -757,7 +757,7 @@ public class ClientCnx extends PulsarHandler {
// verbatim inline continuation.
resolution.whenComplete(guarded);
} else {
- // PIP-478: an AsyncAuthenticationDriver resolved this credential off the event loop; a hung
+ // PIP-478: an BinaryAuthenticationDriver resolved this credential off the event loop; a hung
// driver (getAuthDataAsync / authenticateAsync never completing) would otherwise wedge the connect
// handshake. Bound it with the client's operation-timeout budget: on expiry the resolution is
// completed with a mapped v4 auth exception, and the guarded continuation (still hopped to
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index 873069c1c930..ad0f2cbbe924 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -47,8 +47,8 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -579,9 +579,15 @@ public class PulsarClientImpl implements PulsarClient {
*/
private synchronized Executor blockingAuthExecutor() {
if (blockingAuthExecutor == null) {
- blockingAuthExecutor = new ThreadPoolExecutor(0, AUTH_BLOCKING_MAX_THREADS, 60L, TimeUnit.SECONDS,
- new SynchronousQueue<>(),
+ // Queue rather than reject. Every caller is a connection attempt or an authenticated request, so
+ // a saturated pool must slow them down, not fail them: a SynchronousQueue with the default abort
+ // policy turns a reconnect storm against a slow identity provider into failed connections. Core
+ // threads are allowed to time out, so an unused pool still costs nothing.
+ ThreadPoolExecutor executor = new ThreadPoolExecutor(AUTH_BLOCKING_MAX_THREADS,
+ AUTH_BLOCKING_MAX_THREADS, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(),
new ExecutorProvider.ExtendedThreadFactory("pulsar-client-auth-blocking", true));
+ executor.allowCoreThreadTimeOut(true);
+ blockingAuthExecutor = executor;
}
return blockingAuthExecutor;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/AsyncHttpAuthenticationProvider.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/AsyncHttpAuthenticationProvider.java
index b2c21a193882..efdb08560e1f 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/AsyncHttpAuthenticationProvider.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/AsyncHttpAuthenticationProvider.java
@@ -25,7 +25,7 @@ import java.util.Optional;
* v5-native HTTP multi-round authentication to the framework's HTTP client APIs (PIP-478).
*
* <p>This mirrors, on the HTTP transport, what
- * {@code org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver} does on the binary transport:
+ * {@code org.apache.pulsar.client.api.internal.BinaryAuthenticationDriver} does on the binary transport:
* the HTTP callers ({@code HttpClient} for topic lookup, {@code BaseResource}/{@code ComponentResource}
* for the admin client) test a plugin for this marker and, when present, route the SASL-style
* {@code 401}→resubmit→{@code 200} exchange through the shared {@link HttpAuthenticationDriver} instead of
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/AsyncAuthenticationDriver.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationDriver.java
similarity index 84%
rename from pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/AsyncAuthenticationDriver.java
rename to pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationDriver.java
index 6038a46d0850..ff646e6d7310 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/internal/AsyncAuthenticationDriver.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationDriver.java
@@ -16,20 +16,18 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.client.api.internal;
+package org.apache.pulsar.client.impl.auth.v5;
import java.util.concurrent.CompletableFuture;
import org.apache.pulsar.common.api.AuthData;
/**
- * Stable-internal marker that lets {@code ClientCnx} drive authentication asynchronously (PIP-478).
+ * How {@code ClientCnx} drives authentication over the binary protocol (PIP-478).
*
- * <p>This is the only carve-out from the otherwise-untouched v4 client. When the configured
- * {@link org.apache.pulsar.client.api.Authentication} instance also implements this interface,
- * {@code ClientCnx} routes connect / refresh / challenge handling through these non-blocking methods
- * and pipes the result back onto the Netty event loop; plain v4 instances keep the existing
- * synchronous path verbatim. It is generic across all challenge types (initial connect, the broker's
- * {@code REFRESH_AUTH_DATA} sentinel, SASL multi-round, and custom challenge/response).
+ * <p>The client resolves one v5 {@code Authentication} and wraps it in a driver, from which every connection
+ * attempt opens its own exchange. There is no synchronous alternative: all challenge types (initial connect,
+ * the broker's {@code REFRESH_AUTH_DATA} sentinel, SASL multi-round, and custom challenge/response) run
+ * through these non-blocking methods, with the result piped back onto the Netty event loop.
*
* <p><b>Exchange-scoped.</b> Authentication is driven per <em>exchange</em> — one connection attempt.
* {@code ClientCnx} calls {@link #newAuthenticationExchange(String)} once per connect and drives that
@@ -44,10 +42,10 @@ import org.apache.pulsar.common.api.AuthData;
* conversation state does NOT survive a REFRESH. A fresh exchange is likewise created for every connection
* attempt, so concurrent handshakes to different brokers never collide.
*
- * <p>The {@code .internal.} subpackage signals "stable internal" — application code should not
- * implement this interface; it is observed by the framework.
+ * <p>This lives in the client implementation, not the API module: the client resolves the authentication it
+ * drives itself, so nothing outside {@code pulsar-client} implements or observes this type.
*/
-public interface AsyncAuthenticationDriver {
+public interface BinaryAuthenticationDriver {
/**
* Begin an authentication exchange for a single connection attempt to the given broker.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchange.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchange.java
index 8fdb47b50d70..46456bace241 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchange.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchange.java
@@ -24,12 +24,12 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.auth.AuthChallenge;
import org.apache.pulsar.client.api.v5.auth.Authentication;
import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthChallengeHandler;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.common.api.AuthData;
/**
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5BinaryAuthenticationDriver.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5BinaryAuthenticationDriver.java
index 7421926f2398..9641699129cd 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5BinaryAuthenticationDriver.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/V5BinaryAuthenticationDriver.java
@@ -19,7 +19,6 @@
package org.apache.pulsar.client.impl.auth.v5;
import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver;
import org.apache.pulsar.client.api.v5.auth.Authentication;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServices;
@@ -27,7 +26,7 @@ import org.apache.pulsar.common.api.AuthData;
/**
* Drives a v5-native {@link Authentication} body over the Pulsar binary transport, exposing it as the
- * legacy {@link AsyncAuthenticationDriver} that {@code ClientCnx} consumes (PIP-478). This is
+ * legacy {@link BinaryAuthenticationDriver} that {@code ClientCnx} consumes (PIP-478). This is
* the single exchange pattern shared by the built-in v4 auth shims (Token, Basic, OAuth2, Athenz, SASL);
* each shim wraps its v5 body in this driver rather than re-implementing the exchange.
*
@@ -37,7 +36,7 @@ import org.apache.pulsar.common.api.AuthData;
* binary rules. One {@link BinaryAuthenticationExchange} owns one call context (and its state slot) for
* the lifetime of a single connection attempt, so multi-round conversation state survives across rounds.
*/
-public final class V5BinaryAuthenticationDriver implements AsyncAuthenticationDriver {
+public final class V5BinaryAuthenticationDriver implements BinaryAuthenticationDriver {
private final Authentication v5;
private final String clientInstanceId;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
index b65d892b80c4..1d18d15ac2c3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/v5/package-info.java
@@ -19,7 +19,7 @@
/**
* v5-native implementations of the built-in credential-fetching authentication plugins (PIP-478), and
* the shared {@link org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver} that exposes
- * them to {@code ClientCnx} through the {@code AsyncAuthenticationDriver} carve-out. The v4 plugin
+ * them to {@code ClientCnx} through the {@code BinaryAuthenticationDriver} carve-out. The v4 plugin
* classes in the parent package are thin shims that keep their verbatim synchronous surface and drive
* these bodies on the async binary path.
*/
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
index bbe9999045a5..f3f72142f957 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
@@ -43,8 +43,8 @@ import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.ProxyProtocol;
import org.apache.pulsar.client.api.ServiceUrlProvider;
import org.apache.pulsar.client.api.Socks5ProxyScope;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver;
import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver;
import org.apache.pulsar.client.util.Secret;
import org.apache.pulsar.tls.PulsarTlsFactory;
import org.apache.pulsar.tls.TlsPolicy;
@@ -267,7 +267,7 @@ public class ClientConfigurationData implements Serializable, Cloneable {
// from which ClientCnx opens one exchange per connection attempt. Resolved by PulsarClientImpl; null
// until then (and for a configuration that never builds a client).
@JsonIgnore
- private transient AsyncAuthenticationDriver v5AuthenticationDriver;
+ private transient BinaryAuthenticationDriver v5AuthenticationDriver;
@Schema(
name = "concurrentLookupRequest",
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxAsyncAuthTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxAsyncAuthTest.java
index 5c25efbfc592..9e1293edbe9d 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxAsyncAuthTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxAsyncAuthTest.java
@@ -49,10 +49,10 @@ import java.util.function.Supplier;
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.api.internal.AsyncAuthenticationDriver;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.internal.V5AuthenticationProvider;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.impl.auth.v5.V5AuthenticationLoader;
import org.apache.pulsar.client.impl.auth.v5.V5BinaryAuthenticationDriver;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
@@ -67,7 +67,7 @@ import org.testng.annotations.Test;
/**
* PIP-478: verifies the {@code ClientCnx} async authentication carve-out against an
- * {@link AsyncAuthenticationDriver}-capable plugin. Guards the acceptance invariants of the
+ * {@link BinaryAuthenticationDriver}-capable plugin. Guards the acceptance invariants of the
* single-continuation refactor — the connection's {@code authenticationDataProvider} observable
* ({@code getCommandData()}) reflects the freshly-resolved credential on connect and on a broker-pushed
* REFRESH, a REFRESH does not disconnect ({@code getLastDisconnectedTimestamp()} unchanged), and a
@@ -126,7 +126,7 @@ public class ClientCnxAsyncAuthTest {
conf.setAuthentication(auth);
// PIP-478: the client drives the v5 model, so ClientCnx takes its exchanges from the resolved
// authentication driver rather than from the v4 plugin in conf.
- conf.setV5AuthenticationDriver(auth instanceof AsyncAuthenticationDriver driver ? driver
+ conf.setV5AuthenticationDriver(auth instanceof BinaryAuthenticationDriver driver ? driver
: new V5BinaryAuthenticationDriver(V5AuthenticationLoader.forStartedV4Plugin(auth)));
ClientCnx cnx = new ClientCnx(InstrumentProvider.NOOP, conf, eventLoop);
cnx.setRemoteHostName("localhost");
@@ -465,13 +465,13 @@ public class ClientCnxAsyncAuthTest {
}
/**
- * A minimal {@link AsyncAuthenticationDriver}-capable v4 {@link Authentication} whose exchanges hand
+ * A minimal {@link BinaryAuthenticationDriver}-capable v4 {@link Authentication} whose exchanges hand
* back futures the test completes explicitly (from an off-event-loop thread), so the harness can drive
* the pending / stale / post-close continuation paths deterministically.
*/
// Implements the deprecated v4 Authentication SPI (configure(Map)) by design, like the real plugins.
@SuppressWarnings("deprecation")
- private static final class ControllableAuthDriver implements Authentication, AsyncAuthenticationDriver {
+ private static final class ControllableAuthDriver implements Authentication, BinaryAuthenticationDriver {
private static final long serialVersionUID = 1L;
@@ -526,13 +526,13 @@ public class ClientCnxAsyncAuthTest {
}
/**
- * An {@link AsyncAuthenticationDriver} whose exchanges always resolve inline with a non-terminal
+ * An {@link BinaryAuthenticationDriver} whose exchanges always resolve inline with a non-terminal
* credential — so it answers every challenge and never completes the handshake. Used to drive the
* challenge-round cap deterministically (each round resolves synchronously, so rounds are not dropped by
* the serialize-or-drop guard).
*/
@SuppressWarnings("deprecation")
- private static final class AlwaysRespondingDriver implements Authentication, AsyncAuthenticationDriver {
+ private static final class AlwaysRespondingDriver implements Authentication, BinaryAuthenticationDriver {
private static final long serialVersionUID = 1L;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/AuthInitializationRetryTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/AuthInitializationRetryTest.java
index f6fc6b208044..2fbbf1ce6fdc 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/AuthInitializationRetryTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/AuthInitializationRetryTest.java
@@ -24,12 +24,12 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.auth.Authentication;
import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
import org.apache.pulsar.client.api.v5.auth.AuthenticationInitContext;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthData;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.testng.annotations.Test;
/**
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchangeTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchangeTest.java
index ea42d12bb971..ff6b6bc54d73 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchangeTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/BinaryAuthenticationExchangeTest.java
@@ -21,7 +21,6 @@ package org.apache.pulsar.client.impl.auth.v5;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.auth.AuthChallenge;
import org.apache.pulsar.client.api.v5.auth.Authentication;
import org.apache.pulsar.client.api.v5.auth.AuthenticationCallContext;
@@ -30,6 +29,7 @@ import org.apache.pulsar.client.api.v5.auth.BinaryAuthChallengeHandler;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthData;
import org.apache.pulsar.client.api.v5.auth.BinaryAuthDataProvider;
import org.apache.pulsar.client.api.v5.auth.ChallengeResponse;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.common.api.AuthData;
import org.testng.annotations.Test;
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/CredentialOffloadTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/CredentialOffloadTest.java
index c888bf2e3e01..40ee78574c6f 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/CredentialOffloadTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/auth/v5/CredentialOffloadTest.java
@@ -31,9 +31,9 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicReference;
-import org.apache.pulsar.client.api.internal.AsyncAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.client.api.v5.internal.ClientAuthenticationServices;
import org.apache.pulsar.client.impl.auth.oauth2.OAuth2AuthenticationV5;
+import org.apache.pulsar.client.impl.auth.v5.BinaryAuthenticationDriver.AuthenticationExchange;
import org.apache.pulsar.common.api.AuthData;
import org.testng.annotations.Test;