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/10/17 06:12:20 UTC
(pulsar) 01/14: [fix][client] Surface V5 client failures as the typed exception of their error category (#26950)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9e5578c7f193e2892f9f53824f451cb7b6fb7d16
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Oct 16 16:16:34 2026 -0700
[fix][client] Surface V5 client failures as the typed exception of their error category (#26950)
(cherry picked from commit cb89efaed4f9a84de6e9b86a523dbc0305a0ef7d)
---
.../pulsar/client/api/v5/V5AuthErrorTest.java | 191 +++++++++++++++++++++
.../client/api/v5/V5ProducerAccessModeTest.java | 20 +--
.../api/v5/V5ScalableTopicAutoCreateTest.java | 6 +
.../client/api/v5/V5SchemaEvolutionTest.java | 17 +-
.../pulsar/client/api/v5/V5TransactionTest.java | 43 +++++
.../impl/v5/CheckpointConsumerBuilderV5.java | 11 +-
.../pulsar/client/impl/v5/DagWatchClient.java | 2 +-
.../client/impl/v5/MultiTopicQueueConsumer.java | 2 +-
.../client/impl/v5/MultiTopicStreamConsumer.java | 2 +-
.../pulsar/client/impl/v5/ProducerBuilderV5.java | 12 +-
.../client/impl/v5/PulsarClientBuilderV5.java | 4 +-
.../client/impl/v5/PulsarClientProviderV5.java | 4 +-
.../pulsar/client/impl/v5/PulsarClientV5.java | 13 +-
.../client/impl/v5/QueueConsumerBuilderV5.java | 11 +-
.../client/impl/v5/ScalableCheckpointConsumer.java | 5 +-
.../client/impl/v5/ScalableQueueConsumer.java | 5 +-
.../client/impl/v5/ScalableStreamConsumer.java | 5 +-
.../client/impl/v5/ScalableTopicProducer.java | 52 +++---
.../client/impl/v5/ScalableTopicsWatcher.java | 2 +-
.../client/impl/v5/StreamConsumerBuilderV5.java | 11 +-
.../pulsar/client/impl/v5/TransactionV5.java | 10 +-
.../pulsar/client/impl/v5/V5ExceptionMappings.java | 160 +++++++++++++++++
.../pulsar/client/impl/v5/DagWatchClientTest.java | 26 +++
.../client/impl/v5/PulsarClientBuilderV5Test.java | 13 +-
.../client/impl/v5/V5ExceptionMappingsTest.java | 135 +++++++++++++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 2 +-
26 files changed, 648 insertions(+), 116 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AuthErrorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AuthErrorTest.java
new file mode 100644
index 000000000000..2e802472a2c4
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5AuthErrorTest.java
@@ -0,0 +1,191 @@
+/*
+ * 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.api.v5;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import io.jsonwebtoken.Jwts;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.v5.auth.AuthenticationFactory;
+import org.apache.pulsar.client.api.v5.config.ProducerAccessMode;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
+import org.apache.pulsar.common.policies.data.AuthAction;
+import org.apache.pulsar.security.MockedPulsarStandalone;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * A V5 client whose role is authenticated but not permitted gets a typed
+ * {@link PulsarClientException.AuthorizationException}, whichever request the broker rejects: the topic
+ * lookup (DAG watch), the namespace watch, the subscribe to the topic's controller, or a per-segment producer,
+ * consumer or reader. A client whose credentials are rejected gets a typed
+ * {@link PulsarClientException.AuthenticationException}. See error-model.md §2 and §4.
+ */
+public class V5AuthErrorTest extends MockedPulsarStandalone {
+
+ private static final String NAMESPACE = "public/default";
+ private static final String PRODUCER_ROLE = "v5-producer";
+ private static final String CONSUMER_ROLE = "v5-consumer";
+
+ private PulsarAdmin superUserAdmin;
+ /** No permission at all. */
+ private PulsarClient nobodyClient;
+ /** Produce permission on the namespace only. */
+ private PulsarClient producerClient;
+ /** Consume permission on the namespace only. */
+ private PulsarClient consumerClient;
+
+ @BeforeClass(alwaysRun = true)
+ public void setup() throws Exception {
+ configureTokenAuthentication();
+ configureDefaultAuthorization();
+ start();
+ superUserAdmin = PulsarAdmin.builder()
+ .serviceHttpUrl(getPulsarService().getWebServiceAddress())
+ .authentication(new AuthenticationToken(SUPER_USER_TOKEN))
+ .build();
+ superUserAdmin.namespaces().grantPermissionOnNamespace(NAMESPACE, PRODUCER_ROLE, Set.of(AuthAction.produce));
+ superUserAdmin.namespaces().grantPermissionOnNamespace(NAMESPACE, CONSUMER_ROLE, Set.of(AuthAction.consume));
+ nobodyClient = newClient(NOBODY_TOKEN);
+ producerClient = newClient(tokenOf(PRODUCER_ROLE));
+ consumerClient = newClient(tokenOf(CONSUMER_ROLE));
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void cleanup() throws Exception {
+ for (AutoCloseable c : new AutoCloseable[] {nobodyClient, producerClient, consumerClient, superUserAdmin}) {
+ if (c != null) {
+ c.close();
+ }
+ }
+ close();
+ }
+
+ @Test
+ public void testLookupNotAuthorized() throws Exception {
+ String topic = newScalableTopic();
+
+ assertNotAuthorized(() -> nobodyClient.newProducer(Schema.string()).topic(topic).create());
+ assertNotAuthorized(nobodyClient.newProducer(Schema.string()).topic(topic).createAsync());
+ assertNotAuthorized(() -> nobodyClient.newQueueConsumer(Schema.string()).topic(topic)
+ .subscriptionName("queue").subscribe());
+ assertNotAuthorized(() -> nobodyClient.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribe());
+ assertNotAuthorized(nobodyClient.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribeAsync());
+ assertNotAuthorized(() -> nobodyClient.newCheckpointConsumer(Schema.string()).topic(topic).create());
+ assertNotAuthorized(() -> nobodyClient.newCheckpointConsumer(Schema.string()).topic(topic)
+ .consumerGroup("group").create());
+ }
+
+ @Test
+ public void testNamespaceWatchNotAuthorized() {
+ assertNotAuthorized(() -> nobodyClient.newStreamConsumer(Schema.string()).namespace(NAMESPACE)
+ .subscriptionName("stream").subscribe());
+ assertNotAuthorized(() -> nobodyClient.newQueueConsumer(Schema.string()).namespace(NAMESPACE)
+ .subscriptionName("queue").subscribe());
+ }
+
+ @Test
+ public void testSubscribeNotAuthorized() throws Exception {
+ String topic = newScalableTopic();
+ // The role can look the topic up and produce to it...
+ @Cleanup
+ Producer<String> producer = producerClient.newProducer(Schema.string()).topic(topic).create();
+ producer.newMessage().value("allowed").send();
+
+ // ...but neither the controller (stream, consumer group) nor a segment (queue, checkpoint) lets it consume.
+ assertNotAuthorized(() -> producerClient.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribe());
+ assertNotAuthorized(() -> producerClient.newCheckpointConsumer(Schema.string()).topic(topic)
+ .consumerGroup("group").create());
+ assertNotAuthorized(() -> producerClient.newQueueConsumer(Schema.string()).topic(topic)
+ .subscriptionName("queue").subscribe());
+ assertNotAuthorized(producerClient.newQueueConsumer(Schema.string()).topic(topic)
+ .subscriptionName("queue").subscribeAsync());
+ assertNotAuthorized(() -> producerClient.newCheckpointConsumer(Schema.string()).topic(topic).create());
+ }
+
+ @Test
+ public void testProduceNotAuthorized() throws Exception {
+ String topic = newScalableTopic();
+ // The role can look the topic up and subscribe to it...
+ @Cleanup
+ StreamConsumer<String> consumer = consumerClient.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribe();
+ // ...and create a producer, which attaches to the segments on the first send...
+ @Cleanup
+ Producer<String> producer = consumerClient.newProducer(Schema.string()).topic(topic).create();
+
+ // ...where the segment rejects it.
+ assertNotAuthorized(() -> producer.newMessage().value("rejected").send());
+ assertNotAuthorized(producer.async().newMessage().value("rejected").send());
+ // An exclusive producer attaches to the segments on create.
+ assertNotAuthorized(() -> consumerClient.newProducer(Schema.string()).topic(topic)
+ .accessMode(ProducerAccessMode.EXCLUSIVE).create());
+ }
+
+ @Test
+ public void testCredentialsRejected() throws Exception {
+ String topic = newScalableTopic();
+ @Cleanup
+ PulsarClient client = newClient("not-a-valid-token");
+
+ assertThatThrownBy(() -> client.newProducer(Schema.string()).topic(topic).create())
+ .isInstanceOf(PulsarClientException.AuthenticationException.class);
+ assertThatThrownBy(() -> client.newStreamConsumer(Schema.string()).topic(topic)
+ .subscriptionName("stream").subscribe())
+ .isInstanceOf(PulsarClientException.AuthenticationException.class);
+ }
+
+ private static void assertNotAuthorized(ThrowingCallable call) {
+ assertThatThrownBy(call).isInstanceOf(PulsarClientException.AuthorizationException.class);
+ }
+
+ private static void assertNotAuthorized(CompletableFuture<?> future) {
+ assertThat(future).failsWithin(30, TimeUnit.SECONDS)
+ .withThrowableThat()
+ .withCauseInstanceOf(PulsarClientException.AuthorizationException.class);
+ }
+
+ private String newScalableTopic() throws Exception {
+ String topic = "topic://" + NAMESPACE + "/" + UUID.randomUUID();
+ superUserAdmin.scalableTopics().createScalableTopic(topic, 1);
+ return topic;
+ }
+
+ private PulsarClient newClient(String token) throws PulsarClientException {
+ return PulsarClient.builder()
+ .serviceUrl(getPulsarService().getBrokerServiceUrl())
+ .authentication(AuthenticationFactory.token(token))
+ .build();
+ }
+
+ private static String tokenOf(String role) {
+ return Jwts.builder().claim("sub", role).signWith(SECRET_KEY).compact();
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
index 7424ae30fa2f..b4b466ab97fa 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ProducerAccessModeTest.java
@@ -19,8 +19,7 @@
package org.apache.pulsar.client.api.v5;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.expectThrows;
import java.time.Duration;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.ProducerAccessMode;
@@ -84,18 +83,11 @@ public class V5ProducerAccessModeTest extends V5ClientBaseTest {
// Second EXCLUSIVE on the same topic must fail at create() — the V5 builder
// eagerly claims every active segment up front for exclusive access modes.
- try {
- v5Client.newProducer(Schema.string())
- .topic(topic)
- .accessMode(ProducerAccessMode.EXCLUSIVE)
- .create();
- fail("a second EXCLUSIVE producer should not be able to attach");
- } catch (PulsarClientException expected) {
- String msg = expected.getMessage() == null ? "" : expected.getMessage();
- assertTrue(msg.contains("Busy") || msg.contains("Fenced")
- || msg.contains("ProducerFenced") || msg.contains("ProducerBusy"),
- "unexpected error message for second EXCLUSIVE producer: " + msg);
- }
+ expectThrows(PulsarClientException.ProducerBusyException.class, () ->
+ v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .accessMode(ProducerAccessMode.EXCLUSIVE)
+ .create());
// The first producer is still healthy and can send.
first.newMessage().value("still-here").send();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
index a9c31ebdde9c..1c8493ec71d7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5ScalableTopicAutoCreateTest.java
@@ -99,6 +99,12 @@ public class V5ScalableTopicAutoCreateTest extends V5ClientBaseTest {
// topic does — not a generic PulsarClientException.
assertThrows(PulsarClientException.NotFoundException.class, () ->
v5Client.newProducer(Schema.string()).topic(topic).create());
+ assertThrows(PulsarClientException.NotFoundException.class, () ->
+ v5Client.newQueueConsumer(Schema.string()).topic(topic).subscriptionName("sub").subscribe());
+ assertThrows(PulsarClientException.NotFoundException.class, () ->
+ v5Client.newStreamConsumer(Schema.string()).topic(topic).subscriptionName("sub").subscribe());
+ assertThrows(PulsarClientException.NotFoundException.class, () ->
+ v5Client.newCheckpointConsumer(Schema.string()).topic(topic).create());
} finally {
admin.namespaces().removeAutoTopicCreation(getNamespace());
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SchemaEvolutionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SchemaEvolutionTest.java
index 7c8ac2d669ee..86aba2345792 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SchemaEvolutionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5SchemaEvolutionTest.java
@@ -21,10 +21,9 @@ package org.apache.pulsar.client.api.v5;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.expectThrows;
import java.time.Duration;
import java.util.Objects;
-import java.util.concurrent.CompletionException;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.v5.schema.Schema;
@@ -178,17 +177,9 @@ public class V5SchemaEvolutionTest extends V5ClientBaseTest {
Producer<VBad> bad = v5Client.newProducer(Schema.avro(VBad.class))
.topic(topic)
.create();
- try {
- bad.newMessage().key("bad").value(new VBad(5, 50)).send();
- fail("expected incompatible schema to be rejected post-split");
- } catch (PulsarClientException expected) {
- // good — the broker preserved v1 history across the split and rejected
- // the registration.
- } catch (CompletionException ce) {
- // Some V5 builders may surface the error wrapped — accept either form.
- assertTrue(ce.getCause() instanceof PulsarClientException,
- "unexpected wrapped error: " + ce);
- }
+ // The broker preserved v1 history across the split and rejects the registration.
+ expectThrows(PulsarClientException.IncompatibleSchemaException.class,
+ () -> bad.newMessage().key("bad").value(new VBad(5, 50)).send());
// The registry still shows just v1 — the failed registration didn't leak through.
var schemas = admin.schemas().getAllSchemas(topic);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
index 59057de8bf36..3e91bfe26ab2 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/v5/V5TransactionTest.java
@@ -22,11 +22,13 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import lombok.Cleanup;
import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.v5.config.TransactionPolicy;
@@ -71,6 +73,47 @@ public class V5TransactionTest extends V5ClientBaseTest {
assertEquals(txn.state(), Transaction.State.ABORTED);
}
+ @Test
+ public void testCommitOfAbortedTransactionFailsWithTransactionConflict() throws Exception {
+ // A transaction that is no longer open cannot be committed: the commit fails with a
+ // typed TransactionConflict, blocking and async alike.
+ PulsarClient client = newTxnClient();
+ Transaction txn = client.newTransaction();
+ txn.abort();
+
+ expectThrows(PulsarClientException.TransactionConflictException.class, txn::commit);
+ ExecutionException e = expectThrows(ExecutionException.class, () -> txn.async().commit().get());
+ assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException, e.toString());
+ }
+
+ @Test
+ public void testSendInAbortedTransactionFailsWithTransactionConflict() throws Exception {
+ // A send in a transaction that is no longer open fails with a typed TransactionConflict,
+ // blocking and async alike.
+ PulsarClient client = newTxnClient();
+ String topic = newScalableTopic(1);
+
+ @Cleanup
+ Producer<String> producer = client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+
+ Transaction txn = client.newTransaction();
+ txn.abort();
+
+ expectThrows(PulsarClientException.TransactionConflictException.class,
+ () -> producer.newMessage().transaction(txn).value("blocking").send());
+ ExecutionException e = expectThrows(ExecutionException.class,
+ () -> producer.async().newMessage().transaction(txn).value("async").send().get());
+ assertTrue(e.getCause() instanceof PulsarClientException.TransactionConflictException, e.toString());
+ }
+
+ @Test
+ public void testNewTransactionWithoutTransactionPolicyFailsWithInvalidConfiguration() {
+ // The shared client has no transaction policy, so it has transactions disabled.
+ expectThrows(PulsarClientException.InvalidConfigurationException.class, v5Client::newTransaction);
+ }
+
@Test
public void testCommitMakesProducedMessagesVisible() throws Exception {
PulsarClient client = newTxnClient();
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/CheckpointConsumerBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/CheckpointConsumerBuilderV5.java
index e87a41231dde..5859cc1b5fad 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/CheckpointConsumerBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/CheckpointConsumerBuilderV5.java
@@ -52,10 +52,7 @@ final class CheckpointConsumerBuilderV5<T> implements CheckpointConsumerBuilder<
try {
return createAsync().join();
} catch (java.util.concurrent.CompletionException e) {
- if (e.getCause() instanceof PulsarClientException pce) {
- throw pce;
- }
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -88,14 +85,16 @@ final class CheckpointConsumerBuilderV5<T> implements CheckpointConsumerBuilder<
// its clean unsubscribe once the in-flight attempt settles.
session.close();
}
- });
+ })
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
// Unmanaged: read every active segment, no broker-side state.
DagWatchClient dagWatch = new DagWatchClient(client.v4Client(), topic);
return dagWatch.start()
.thenCompose(initialLayout -> ScalableCheckpointConsumer.createUnmanagedAsync(
- client, v5Schema, dagWatch, initialLayout, startPosition, consumerName));
+ client, v5Schema, dagWatch, initialLayout, startPosition, consumerName))
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/DagWatchClient.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/DagWatchClient.java
index dcb152a24720..dd01b80922ad 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/DagWatchClient.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/DagWatchClient.java
@@ -236,7 +236,7 @@ final class DagWatchClient implements DagWatchSession, AutoCloseable {
initialLayoutFuture.completeExceptionally(
error == ServerError.TopicNotFound
? new PulsarClientException.NotFoundException(detail)
- : new PulsarClientException(detail));
+ : ClientCnx.getPulsarClientException(error, detail));
}
// After the initial layout has arrived, broker-side errors on this session
// (e.g., metadata unavailable) are transient — a reconnect typically clears
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
index b808536d4e02..11cf0a5de578 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicQueueConsumer.java
@@ -375,7 +375,7 @@ final class MultiTopicQueueConsumer<T> implements QueueConsumerImpl<T> {
Thread.currentThread().interrupt();
throw new PulsarClientException("Close interrupted", e);
} catch (ExecutionException e) {
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
index 4b6d40c5eac8..1e4903adc0f6 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/MultiTopicStreamConsumer.java
@@ -381,7 +381,7 @@ final class MultiTopicStreamConsumer<T> implements StreamConsumer<T> {
Thread.currentThread().interrupt();
throw new PulsarClientException("Close interrupted", e);
} catch (ExecutionException e) {
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
index db04b64994a2..c244fe9d970a 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ProducerBuilderV5.java
@@ -58,14 +58,7 @@ final class ProducerBuilderV5<T> implements ProducerBuilder<T> {
try {
return createAsync().join();
} catch (java.util.concurrent.CompletionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof PulsarClientException pce) {
- throw pce;
- }
- if (cause instanceof org.apache.pulsar.client.api.PulsarClientException.NotFoundException) {
- throw new PulsarClientException.NotFoundException(cause.getMessage());
- }
- throw new PulsarClientException(cause);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -94,7 +87,8 @@ final class ProducerBuilderV5<T> implements ProducerBuilder<T> {
throw ex instanceof java.util.concurrent.CompletionException ce
? ce : new java.util.concurrent.CompletionException(ex);
}));
- });
+ })
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5.java
index 94183dfd381c..b6a6c2ca4ed7 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5.java
@@ -117,7 +117,7 @@ final class PulsarClientBuilderV5 implements PulsarClientBuilder {
var v4Client = new PulsarClientImpl(clientConf);
return new PulsarClientV5(v4Client, description, transactionTimeout);
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
- throw new PulsarClientException(e.getMessage(), e);
+ throw V5ExceptionMappings.toV5Exception(e);
} finally {
// Whether the factory was consumed is read off the copy the client was given rather than
// inferred from how the build ended: a build that failed after the framework took the instance
@@ -222,7 +222,7 @@ final class PulsarClientBuilderV5 implements PulsarClientBuilder {
try {
this.v5Authentication = V5AuthenticationLoader.create(authPluginClassName, authParamsString);
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
- throw new PulsarClientException(e.getMessage(), e);
+ throw V5ExceptionMappings.toV5Exception(e);
}
return this;
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientProviderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientProviderV5.java
index eb05fc8dea15..79809a120b26 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientProviderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientProviderV5.java
@@ -194,7 +194,7 @@ public final class PulsarClientProviderV5 implements PulsarClientProvider {
try {
return V5AuthenticationLoader.create(className, params);
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
- throw new PulsarClientException(e.getMessage(), e);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -204,7 +204,7 @@ public final class PulsarClientProviderV5 implements PulsarClientProvider {
try {
return V5AuthenticationLoader.create(className, params);
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
- throw new PulsarClientException(e.getMessage(), e);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
index 792d9ea060ce..599d1fddbd2b 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/PulsarClientV5.java
@@ -20,7 +20,6 @@ package org.apache.pulsar.client.impl.v5;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.v5.CheckpointConsumerBuilder;
@@ -81,8 +80,7 @@ final class PulsarClientV5 implements PulsarClient {
try {
return newTransactionAsync().get();
} catch (ExecutionException e) {
- Throwable cause = e.getCause() != null ? e.getCause() : e;
- throw new PulsarClientException(cause.getMessage(), cause);
+ throw V5ExceptionMappings.toV5Exception(e);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new PulsarClientException("Interrupted while creating transaction", e);
@@ -95,7 +93,8 @@ final class PulsarClientV5 implements PulsarClient {
if (transactionTimeout != null) {
builder.withTransactionTimeout(transactionTimeout.toMillis(), TimeUnit.MILLISECONDS);
}
- return builder.build().thenApply(v4Txn -> (Transaction) new TransactionV5(v4Txn));
+ return builder.build().thenApply(v4Txn -> (Transaction) new TransactionV5(v4Txn))
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
@@ -103,15 +102,13 @@ final class PulsarClientV5 implements PulsarClient {
try {
v4Client.close();
} catch (org.apache.pulsar.client.api.PulsarClientException e) {
- throw new PulsarClientException(e.getMessage(), e);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@Override
public CompletableFuture<Void> closeAsync() {
- return v4Client.closeAsync().exceptionally(ex -> {
- throw new CompletionException(new PulsarClientException(ex.getMessage(), ex));
- });
+ return v4Client.closeAsync().exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/QueueConsumerBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/QueueConsumerBuilderV5.java
index c1d8bdf261fb..b24626d5720e 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/QueueConsumerBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/QueueConsumerBuilderV5.java
@@ -66,10 +66,7 @@ final class QueueConsumerBuilderV5<T> implements QueueConsumerBuilder<T> {
try {
return subscribeAsync().join();
} catch (java.util.concurrent.CompletionException e) {
- if (e.getCause() instanceof PulsarClientException pce) {
- throw pce;
- }
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -89,13 +86,15 @@ final class QueueConsumerBuilderV5<T> implements QueueConsumerBuilder<T> {
if (namespaceSet) {
return MultiTopicQueueConsumer.createAsync(
- client, v5Schema, conf, namespaceName, propertyFilters);
+ client, v5Schema, conf, namespaceName, propertyFilters)
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
TopicName topic = V5Utils.parseScalableTopicInput(topicName);
DagWatchClient dagWatch = new DagWatchClient(client.v4Client(), topic);
return dagWatch.start()
.thenCompose(initialLayout -> ScalableQueueConsumer.createAsync(
- client, v5Schema, conf, dagWatch, initialLayout, dlqPolicy));
+ client, v5Schema, conf, dagWatch, initialLayout, dlqPolicy))
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
index 7ce0b1ad694f..8e434c2f8000 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableCheckpointConsumer.java
@@ -277,7 +277,7 @@ final class ScalableCheckpointConsumer<T> implements CheckpointConsumer<T> {
Thread.currentThread().interrupt();
throw new PulsarClientException("Close interrupted", e);
} catch (ExecutionException e) {
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -319,7 +319,8 @@ final class ScalableCheckpointConsumer<T> implements CheckpointConsumer<T> {
: CompletableFuture.completedFuture(null)));
}
return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
- .whenComplete((__, ___) -> segmentReaders.clear());
+ .whenComplete((__, ___) -> segmentReaders.clear())
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
// --- Assignment change handling ---
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
index 02e40c15078d..2e6623759ee5 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableQueueConsumer.java
@@ -277,7 +277,7 @@ final class ScalableQueueConsumer<T> implements QueueConsumerImpl<T>, DagWatchCl
Thread.currentThread().interrupt();
throw new PulsarClientException("Close interrupted", e);
} catch (ExecutionException e) {
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -309,7 +309,8 @@ final class ScalableQueueConsumer<T> implements QueueConsumerImpl<T>, DagWatchCl
: CompletableFuture.completedFuture(null)));
}
return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
- .whenComplete((__, ___) -> segmentConsumers.clear());
+ .whenComplete((__, ___) -> segmentConsumers.clear())
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
// --- Layout change handling ---
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
index 9764b27876c0..2ff846b67c46 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableStreamConsumer.java
@@ -467,7 +467,7 @@ final class ScalableStreamConsumer<T>
Thread.currentThread().interrupt();
throw new PulsarClientException("Close interrupted", e);
} catch (ExecutionException e) {
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -515,7 +515,8 @@ final class ScalableStreamConsumer<T>
sharedSegmentUnacked.clear();
lastCumulativeAcked.clear();
cumulativeAckWatermark.clear();
- });
+ })
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
// --- Assignment change handling ---
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
index 62e468edfce4..9d5010d8b259 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicProducer.java
@@ -431,18 +431,13 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
var v4MsgId = sendV4(producer, send);
return new MessageIdV5(v4MsgId, segmentId);
} catch (PulsarClientException e) {
- // Thrown while (re)creating the per-segment producer — already a V5 exception
- // (it may wrap a v4 TopicTerminated/AlreadyClosed cause).
+ // Thrown while (re)creating the per-segment producer or by its send(): the V5
+ // exception for the v4 failure, which is its cause (a v4 TopicTerminated or
+ // AlreadyClosed one when the segment is gone).
if (!isSegmentGoneError(e)) {
throw e;
}
lastError = e;
- } catch (org.apache.pulsar.client.api.PulsarClientException e) {
- // Thrown by the v4 producer's send().
- if (!isSegmentGoneError(e)) {
- throw new PulsarClientException(e.getMessage(), e);
- }
- lastError = new PulsarClientException(e.getMessage(), e);
}
// The target segment is gone: sealed by a split/merge, or terminated by a
// regular-to-scalable migration. Wait for the DAG watch to deliver the new
@@ -471,29 +466,33 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
/**
* The v4 {@code send()} with the payload accounting handed over in between its steps: enqueue,
* drop this layer's share of the payload, flush so that a batched message does not wait out the
- * batching delay, then wait for the acknowledgement.
+ * batching delay, then wait for the acknowledgement. It fails with the V5 exception for the
+ * send's own failure, as the async send does, not for its v4 {@code PulsarClientException.unwrap},
+ * which turns a failure of another type, such as a transaction that is no longer open, into a
+ * plain v4 exception.
*/
private org.apache.pulsar.client.api.MessageId sendV4(org.apache.pulsar.client.api.Producer<T> producer,
- PendingSend<T> send)
- throws org.apache.pulsar.client.api.PulsarClientException {
+ PendingSend<T> send) throws PulsarClientException {
+ var ackFuture = buildV4Message(producer, send).sendAsync();
+ if (!ackFuture.isDone()) {
+ releasePayloadShare(send);
+ producer.flushAsync();
+ }
try {
- var ackFuture = buildV4Message(producer, send).sendAsync();
- if (!ackFuture.isDone()) {
- releasePayloadShare(send);
- producer.flushAsync();
- }
return ackFuture.get();
- } catch (Exception e) {
- throw org.apache.pulsar.client.api.PulsarClientException.unwrap(e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new PulsarClientException("Interrupted while waiting for the send to be acknowledged", e);
+ } catch (ExecutionException e) {
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
/**
* True if {@code t} (or one of its causes) signals that the target segment is gone —
* sealed by a split/merge or terminated by a regular-to-scalable migration — so the send
- * should be retried once the new layout arrives. Handles both the v4 exceptions thrown by
- * {@code send()} and the V5-wrapped exceptions thrown while (re)creating the per-segment
- * producer on a now-terminated topic.
+ * should be retried once the new layout arrives. Handles both the v4 exceptions that fail an
+ * async send and the V5 exceptions, caused by them, that fail a blocking one.
*/
private static boolean isSegmentGoneError(Throwable t) {
for (Throwable cause = t; cause != null; cause = cause.getCause()) {
@@ -647,7 +646,7 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
.log("Target segment gone, retrying async send after layout update");
retry.run();
} else {
- finish(send, null, ex);
+ finish(send, null, V5ExceptionMappings.toV5Exception(ex));
}
}
@@ -995,14 +994,7 @@ final class ScalableTopicProducer<T> implements Producer<T>, DagWatchClient.Layo
Thread.currentThread().interrupt();
throw new PulsarClientException("Interrupted while creating segment producer", e);
} catch (ExecutionException e) {
- Throwable cause = e.getCause();
- if (cause instanceof org.apache.pulsar.client.api.PulsarClientException v4Exc) {
- throw new PulsarClientException(v4Exc.getMessage(), v4Exc);
- }
- if (cause instanceof PulsarClientException v5Exc) {
- throw v5Exc;
- }
- throw new PulsarClientException(cause != null ? cause : e);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
index 984a55a8afd6..21755ddbf566 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
@@ -298,7 +298,7 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
.log("WatchScalableTopics rejected");
if (!initialSnapshotFuture.isDone()) {
initialSnapshotFuture.completeExceptionally(
- new PulsarClientException("WatchScalableTopics failed: " + error
+ ClientCnx.getPulsarClientException(error, "WatchScalableTopics failed: " + error
+ (message != null ? " - " + message : "")));
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java
index a408fdf9fa69..d4dab5d38c16 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/StreamConsumerBuilderV5.java
@@ -56,10 +56,7 @@ final class StreamConsumerBuilderV5<T> implements StreamConsumerBuilder<T> {
try {
return subscribeAsync().join();
} catch (java.util.concurrent.CompletionException e) {
- if (e.getCause() instanceof PulsarClientException pce) {
- throw pce;
- }
- throw new PulsarClientException(e.getCause());
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -84,7 +81,8 @@ final class StreamConsumerBuilderV5<T> implements StreamConsumerBuilder<T> {
if (namespaceSet) {
return MultiTopicStreamConsumer.createAsync(
- client, v5Schema, conf, namespaceName, propertyFilters);
+ client, v5Schema, conf, namespaceName, propertyFilters)
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
TopicName topic = V5Utils.parseScalableTopicInput(topicName);
@@ -107,7 +105,8 @@ final class StreamConsumerBuilderV5<T> implements StreamConsumerBuilder<T> {
// path already closed the session.
session.close();
}
- });
+ })
+ .exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/TransactionV5.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/TransactionV5.java
index ff7fa5223c15..2cbe476afa17 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/TransactionV5.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/TransactionV5.java
@@ -50,8 +50,7 @@ final class TransactionV5 implements Transaction {
Thread.currentThread().interrupt();
throw new PulsarClientException("Commit interrupted", e);
} catch (ExecutionException e) {
- Throwable cause = e.getCause() != null ? e.getCause() : e;
- throw new PulsarClientException(cause.getMessage(), cause);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -63,8 +62,7 @@ final class TransactionV5 implements Transaction {
Thread.currentThread().interrupt();
throw new PulsarClientException("Abort interrupted", e);
} catch (ExecutionException e) {
- Throwable cause = e.getCause() != null ? e.getCause() : e;
- throw new PulsarClientException(cause.getMessage(), cause);
+ throw V5ExceptionMappings.toV5Exception(e);
}
}
@@ -103,12 +101,12 @@ final class TransactionV5 implements Transaction {
private final class AsyncView implements AsyncTransaction {
@Override
public CompletableFuture<Void> commit() {
- return v4Transaction.commit();
+ return v4Transaction.commit().exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
@Override
public CompletableFuture<Void> abort() {
- return v4Transaction.abort();
+ return v4Transaction.abort().exceptionallyCompose(V5ExceptionMappings::failWithV5Exception);
}
}
}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappings.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappings.java
new file mode 100644
index 000000000000..d3089ea2b486
--- /dev/null
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappings.java
@@ -0,0 +1,160 @@
+/*
+ * 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.v5;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClientException;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.apache.pulsar.common.util.FutureUtil;
+
+/**
+ * Converts the failures of the operations that the V5 client runs on the v4 client into the V5 exceptions that the
+ * application gets, typed by their error category (error-model.md §2).
+ */
+final class V5ExceptionMappings {
+
+ /** The V5 type of each error category of error-model.md §2, for the v4 exceptions that report it. */
+ private static final List<V5Type> V5_TYPES = List.of(
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.NotFoundException.class,
+ PulsarClientException.NotFoundException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.TopicDoesNotExistException.class,
+ PulsarClientException.NotFoundException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.AuthenticationException.class,
+ PulsarClientException.AuthenticationException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.GettingAuthenticationDataException.class,
+ PulsarClientException.GettingAuthenticationDataException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException.class,
+ PulsarClientException.UnsupportedAuthenticationException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.AuthorizationException.class,
+ PulsarClientException.AuthorizationException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.TimeoutException.class,
+ PulsarClientException.TimeoutException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException.class,
+ PulsarClientException.AlreadyClosedException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ProducerBusyException.class,
+ PulsarClientException.ProducerBusyException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ProducerFencedException.class,
+ PulsarClientException.ProducerBusyException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException.class,
+ PulsarClientException.ConsumerBusyException::new),
+ // Overlapping Key_Shared hash ranges: a subscription constraint that keeps the consumer out.
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ConsumerAssignException.class,
+ PulsarClientException.ConsumerBusyException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.IncompatibleSchemaException.class,
+ PulsarClientException.IncompatibleSchemaException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException.class,
+ PulsarClientException.TopicTerminatedException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ConnectException.class,
+ PulsarClientException.ConnectException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.NotConnectedException.class,
+ PulsarClientException.NotConnectedException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError.class,
+ PulsarClientException.ProducerQueueIsFullException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.MemoryBufferIsFullError.class,
+ PulsarClientException.MemoryBufferIsFullException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.TransactionConflictException.class,
+ PulsarClientException.TransactionConflictException::new),
+ // A transaction with a failed operation, no longer open, or no longer known to its coordinator (timed
+ // out) cannot be committed or aborted (client-api.md).
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.TransactionHasOperationFailedException.class,
+ PulsarClientException.TransactionConflictException::new),
+ new V5Type(TransactionCoordinatorClientException.InvalidTxnStatusException.class,
+ PulsarClientException.TransactionConflictException::new),
+ new V5Type(TransactionCoordinatorClientException.TransactionNotFoundException.class,
+ PulsarClientException.TransactionConflictException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.CryptoException.class,
+ PulsarClientException.CryptoException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.InvalidConfigurationException.class,
+ PulsarClientException.InvalidConfigurationException::new),
+ new V5Type(org.apache.pulsar.client.api.PulsarClientException.InvalidServiceURL.class,
+ PulsarClientException.InvalidServiceURLException::new));
+
+ /** {@link #V5_TYPES} by their v4 type, for a failure of exactly one of these types. */
+ private static final Map<Class<?>, V5Type> V5_TYPES_BY_V4_TYPE = V5_TYPES.stream()
+ .collect(Collectors.toUnmodifiableMap(V5Type::v4Type, Function.identity()));
+
+ /**
+ * {@link #V5_TYPES} with each v4 type before its supertypes, for a failure of a subtype of them: the closest
+ * supertype decides, whatever the order of {@link #V5_TYPES}.
+ */
+ private static final List<V5Type> V5_TYPES_SUBTYPES_FIRST = V5_TYPES.stream()
+ .sorted(Comparator.comparingInt(V5Type::depth).reversed())
+ .toList();
+
+ private V5ExceptionMappings() {
+ }
+
+ /**
+ * The V5 exception to hand the application for the failure of an operation that the V5 client runs on the
+ * v4 client, so that it can branch on the error category (error-model.md §2): the V5 type of its category,
+ * or the plain {@link PulsarClientException} for a failure outside them, either one caused by the original
+ * failure. A V5 exception is returned as is. A {@code CompletionException} or {@code ExecutionException} is
+ * unwrapped first.
+ */
+ static PulsarClientException toV5Exception(Throwable failure) {
+ Throwable cause = FutureUtil.unwrapCompletionException(failure);
+ if (cause instanceof PulsarClientException v5Exception) {
+ return v5Exception;
+ }
+ V5Type type = v5TypeOf(cause);
+ if (type == null) {
+ return new PulsarClientException(cause.getMessage(), cause);
+ }
+ PulsarClientException typed = type.constructor().apply(cause.getMessage());
+ // Unlike the plain exception, most of these types have no constructor that takes the cause.
+ typed.initCause(cause);
+ return typed;
+ }
+
+ /** A future failed with the {@link #toV5Exception V5 exception} for {@code failure}. */
+ static <T> CompletableFuture<T> failWithV5Exception(Throwable failure) {
+ return CompletableFuture.failedFuture(toV5Exception(failure));
+ }
+
+ private static V5Type v5TypeOf(Throwable failure) {
+ V5Type type = V5_TYPES_BY_V4_TYPE.get(failure.getClass());
+ if (type != null) {
+ return type;
+ }
+ for (V5Type candidate : V5_TYPES_SUBTYPES_FIRST) {
+ if (candidate.v4Type().isInstance(failure)) {
+ return candidate;
+ }
+ }
+ return null;
+ }
+
+ /** The V5 exception type, by its message constructor, for the v4 exceptions of {@code v4Type}. */
+ private record V5Type(Class<? extends Exception> v4Type, Function<String, PulsarClientException> constructor) {
+
+ /** How deep {@link #v4Type} is in its class hierarchy: a subtype is deeper than its supertypes. */
+ int depth() {
+ int depth = 0;
+ for (Class<?> type = v4Type; type != null; type = type.getSuperclass()) {
+ depth++;
+ }
+ return depth;
+ }
+ }
+}
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/DagWatchClientTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/DagWatchClientTest.java
index 7f8922312b62..cec11c676fa4 100644
--- a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/DagWatchClientTest.java
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/DagWatchClientTest.java
@@ -30,11 +30,14 @@ import io.netty.util.Timer;
import io.netty.util.TimerTask;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.api.proto.ScalableTopicDAG;
+import org.apache.pulsar.common.api.proto.ServerError;
import org.apache.pulsar.common.naming.TopicName;
import org.testng.annotations.Test;
@@ -142,6 +145,29 @@ public class DagWatchClientTest {
assertThat(changes).containsExactly("1->2", "2->3");
}
+ @Test
+ public void testRejectedLookupFailsStartWithItsErrorCategory() {
+ assertThat(startFailure(ServerError.TopicNotFound))
+ .hasCauseInstanceOf(PulsarClientException.NotFoundException.class);
+ assertThat(startFailure(ServerError.AuthorizationError))
+ .hasCauseInstanceOf(PulsarClientException.AuthorizationException.class);
+ assertThat(startFailure(ServerError.AuthenticationError))
+ .hasCauseInstanceOf(PulsarClientException.AuthenticationException.class);
+ }
+
+ /** How {@link DagWatchClient#start} fails when the broker rejects the lookup with {@code error}. */
+ private static Throwable startFailure(ServerError error) {
+ PulsarClientImpl v4Client = mock(PulsarClientImpl.class);
+ // The broker's answer is fed in by the test: the connection never opens.
+ when(v4Client.getConnection(TOPIC)).thenReturn(new CompletableFuture<>());
+ DagWatchClient watch = new DagWatchClient(v4Client, TopicName.get(TOPIC));
+ CompletableFuture<ClientSegmentLayout> start = watch.start();
+
+ watch.onError(error, "rejected");
+
+ return catchThrowable(start::join);
+ }
+
private static void awaitUninterruptibly(CountDownLatch latch) {
try {
latch.await();
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5Test.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5Test.java
index 25f691b6ba28..dd17fc9236cd 100644
--- a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5Test.java
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/PulsarClientBuilderV5Test.java
@@ -798,13 +798,20 @@ public class PulsarClientBuilderV5Test {
@Test
public void testAuthenticationPluginNotFoundIsWrapped() {
- // A bad plugin class name should surface as V5 PulsarClientException (not a v4 exception
- // type leaking through the surface).
+ // A bad plugin class name should surface as the V5 PulsarClientException of its category (not a
+ // v4 exception type leaking through the surface).
PulsarClientBuilderV5 builder = new PulsarClientBuilderV5();
- assertThrows(PulsarClientException.class, () ->
+ assertThrows(PulsarClientException.UnsupportedAuthenticationException.class, () ->
builder.authentication("com.example.NoSuchAuth", ""));
}
+ @Test
+ public void testBuildWithMalformedServiceUrlFailsWithInvalidServiceUrl() {
+ // The pulsar:// scheme passes the builder's check; the client rejects the port when it is built.
+ PulsarClientBuilder builder = PulsarClient.builder().serviceUrl("pulsar://localhost:not-a-port");
+ assertThrows(PulsarClientException.InvalidServiceURLException.class, builder::build);
+ }
+
/**
* PIP-478: a cross-format TLS-material fold must fail loud, not silently drop trust anchors. A PEM
* {@code tlsPolicy(...)} carrying a private-CA {@code trustCertsFilePath} combined with a keystore-format
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappingsTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappingsTest.java
new file mode 100644
index 000000000000..b7f34056a3b8
--- /dev/null
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/V5ExceptionMappingsTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.v5;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertSame;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClientException;
+import org.apache.pulsar.client.api.v5.PulsarClientException;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class V5ExceptionMappingsTest {
+
+ /** A v4 exception, and the V5 type of its error category (error-model.md §2). */
+ @DataProvider
+ public static Object[][] errorCategories() {
+ return new Object[][] {
+ {new org.apache.pulsar.client.api.PulsarClientException.NotFoundException("error"),
+ PulsarClientException.NotFoundException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.TopicDoesNotExistException("error"),
+ PulsarClientException.NotFoundException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.AuthenticationException("error"),
+ PulsarClientException.AuthenticationException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.GettingAuthenticationDataException("error"),
+ PulsarClientException.GettingAuthenticationDataException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException("error"),
+ PulsarClientException.UnsupportedAuthenticationException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.AuthorizationException("error"),
+ PulsarClientException.AuthorizationException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.TimeoutException("error"),
+ PulsarClientException.TimeoutException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException("error"),
+ PulsarClientException.AlreadyClosedException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ProducerBusyException("error"),
+ PulsarClientException.ProducerBusyException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ProducerFencedException("error"),
+ PulsarClientException.ProducerBusyException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException("error"),
+ PulsarClientException.ConsumerBusyException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ConsumerAssignException("error"),
+ PulsarClientException.ConsumerBusyException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.IncompatibleSchemaException("error"),
+ PulsarClientException.IncompatibleSchemaException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedException("error"),
+ PulsarClientException.TopicTerminatedException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ConnectException("error"),
+ PulsarClientException.ConnectException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.NotConnectedException("error"),
+ PulsarClientException.NotConnectedException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError("error"),
+ PulsarClientException.ProducerQueueIsFullException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.MemoryBufferIsFullError("error"),
+ PulsarClientException.MemoryBufferIsFullException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.TransactionConflictException("error"),
+ PulsarClientException.TransactionConflictException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.TransactionHasOperationFailedException("error"),
+ PulsarClientException.TransactionConflictException.class},
+ {new TransactionCoordinatorClientException.InvalidTxnStatusException("error"),
+ PulsarClientException.TransactionConflictException.class},
+ {new TransactionCoordinatorClientException.TransactionNotFoundException("error"),
+ PulsarClientException.TransactionConflictException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.CryptoException("error"),
+ PulsarClientException.CryptoException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.InvalidConfigurationException("error"),
+ PulsarClientException.InvalidConfigurationException.class},
+ {new org.apache.pulsar.client.api.PulsarClientException.InvalidServiceURL("error", null),
+ PulsarClientException.InvalidServiceURLException.class},
+ };
+ }
+
+ @Test(dataProvider = "errorCategories")
+ public void testToV5ExceptionKeepsTheErrorCategory(Exception v4Exception, Class<?> v5Type) {
+ assertConverted(V5ExceptionMappings.toV5Exception(v4Exception), v5Type, v4Exception);
+ assertConverted(V5ExceptionMappings.toV5Exception(new CompletionException(v4Exception)), v5Type,
+ v4Exception);
+ assertConverted(V5ExceptionMappings.toV5Exception(new ExecutionException(v4Exception)), v5Type,
+ v4Exception);
+ }
+
+ /** A v4 exception type that the mappings don't list. */
+ private static class CustomTimeoutException
+ extends org.apache.pulsar.client.api.PulsarClientException.TimeoutException {
+ CustomTimeoutException(String message) {
+ super(message);
+ }
+ }
+
+ @Test
+ public void testToV5ExceptionTypesASubtypeByItsClosestMappedSupertype() {
+ var timeout = new CustomTimeoutException("timed out");
+
+ assertConverted(V5ExceptionMappings.toV5Exception(timeout), PulsarClientException.TimeoutException.class,
+ timeout);
+ }
+
+ @Test
+ public void testToV5ExceptionWrapsOtherFailuresInThePlainException() {
+ var notAllowed = new org.apache.pulsar.client.api.PulsarClientException.NotAllowedException("not allowed");
+
+ assertConverted(V5ExceptionMappings.toV5Exception(new CompletionException(notAllowed)),
+ PulsarClientException.class, notAllowed);
+ }
+
+ @Test
+ public void testToV5ExceptionReturnsV5ExceptionsAsIs() {
+ var closed = new PulsarClientException.AlreadyClosedException("closed");
+
+ assertSame(V5ExceptionMappings.toV5Exception(new CompletionException(closed)), closed);
+ }
+
+ private static void assertConverted(PulsarClientException converted, Class<?> expectedType,
+ Throwable v4Exception) {
+ assertEquals(converted.getClass(), expectedType);
+ assertEquals(converted.getMessage(), v4Exception.getMessage());
+ assertSame(converted.getCause(), v4Exception);
+ }
+}
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 0015cd7bacc1..120830bf8292 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
@@ -1753,7 +1753,7 @@ public class ClientCnx extends PulsarHandler {
if (cmd.hasError()) {
CompletableFuture<? extends Object> requestFuture = pendingRequests.get(requestId);
if (requestFuture != null && removePendingRequest(requestId, requestFuture)) {
- requestFuture.completeExceptionally(new PulsarClientException(
+ requestFuture.completeExceptionally(getPulsarClientException(cmd.getError(),
"Scalable topic subscribe failed: " + cmd.getError()
+ (cmd.hasMessage() ? " - " + cmd.getMessage() : "")));
}