You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/02/26 10:06:35 UTC
(pulsar) branch branch-3.0 updated: [fix] [client] Do no retrying for error subscription not found when disabled allowAutoSubscriptionCreation (#22078)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new bbf6ddf9244 [fix] [client] Do no retrying for error subscription not found when disabled allowAutoSubscriptionCreation (#22078)
bbf6ddf9244 is described below
commit bbf6ddf9244c1dc02cc157a014c153a09f45be1f
Author: zifengmo <38...@users.noreply.github.com>
AuthorDate: Mon Feb 26 18:06:28 2024 +0800
[fix] [client] Do no retrying for error subscription not found when disabled allowAutoSubscriptionCreation (#22078)
---
.../broker/service/BrokerServiceException.java | 2 ++
.../pulsar/client/api/MultiTopicsConsumerTest.java | 32 ++++++++++++++++++++++
.../pulsar/client/api/PulsarClientException.java | 17 ++++++++++++
.../org/apache/pulsar/client/impl/ClientCnx.java | 2 ++
4 files changed, 53 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
index 3e77588b245..831d6068e20 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerServiceException.java
@@ -258,6 +258,8 @@ public class BrokerServiceException extends Exception {
return ServerError.ServiceNotReady;
} else if (t instanceof TopicNotFoundException) {
return ServerError.TopicNotFound;
+ } else if (t instanceof SubscriptionNotFoundException) {
+ return ServerError.SubscriptionNotFound;
} else if (t instanceof IncompatibleSchemaException
|| t instanceof InvalidSchemaDataException) {
// for backward compatible with old clients, invalid schema data
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
index 315ce378d69..bb8bab29ad9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
@@ -371,4 +371,36 @@ public class MultiTopicsConsumerTest extends ProducerConsumerBase {
assertTrue(consumer instanceof MultiTopicsConsumerImpl);
assertTrue(consumer.isConnected());
}
+
+ @Test(timeOut = 30000)
+ public void testSubscriptionNotFound() throws PulsarAdminException, PulsarClientException {
+ final var topic1 = newTopicName();
+ final var topic2 = newTopicName();
+
+ pulsar.getConfiguration().setAllowAutoSubscriptionCreation(false);
+
+ try {
+ final var singleTopicConsumer = pulsarClient.newConsumer()
+ .topic(topic1)
+ .subscriptionName("sub-1")
+ .isAckReceiptEnabled(true)
+ .subscribe();
+ assertTrue(singleTopicConsumer instanceof ConsumerImpl);
+ } catch (Throwable t) {
+ assertTrue(t.getCause().getCause() instanceof PulsarClientException.SubscriptionNotFoundException);
+ }
+
+ try {
+ final var multiTopicsConsumer = pulsarClient.newConsumer()
+ .topics(List.of(topic1, topic2))
+ .subscriptionName("sub-2")
+ .isAckReceiptEnabled(true)
+ .subscribe();
+ assertTrue(multiTopicsConsumer instanceof MultiTopicsConsumerImpl);
+ } catch (Throwable t) {
+ assertTrue(t.getCause().getCause() instanceof PulsarClientException.SubscriptionNotFoundException);
+ }
+
+ pulsar.getConfiguration().setAllowAutoSubscriptionCreation(true);
+ }
}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index 9409eefe2e0..22a97571e53 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -344,6 +344,22 @@ public class PulsarClientException extends IOException {
}
}
+ /**
+ * Not found subscription that cannot be created.
+ */
+ public static class SubscriptionNotFoundException extends PulsarClientException {
+ /**
+ * Constructs an {@code SubscriptionNotFoundException} with the specified detail message.
+ *
+ * @param msg
+ * The detail message (which is saved for later retrieval
+ * by the {@link #getMessage()} method)
+ */
+ public SubscriptionNotFoundException(String msg) {
+ super(msg);
+ }
+ }
+
/**
* Lookup exception thrown by Pulsar client.
*/
@@ -1159,6 +1175,7 @@ public class PulsarClientException extends IOException {
|| t instanceof NotFoundException
|| t instanceof IncompatibleSchemaException
|| t instanceof TopicDoesNotExistException
+ || t instanceof SubscriptionNotFoundException
|| t instanceof UnsupportedAuthenticationException
|| t instanceof InvalidMessageException
|| t instanceof InvalidTopicNameException
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 f3e8b2354b3..496a9ca0427 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
@@ -1292,6 +1292,8 @@ public class ClientCnx extends PulsarHandler {
return new PulsarClientException.IncompatibleSchemaException(errorMsg);
case TopicNotFound:
return new PulsarClientException.TopicDoesNotExistException(errorMsg);
+ case SubscriptionNotFound:
+ return new PulsarClientException.SubscriptionNotFoundException(errorMsg);
case ConsumerAssignError:
return new PulsarClientException.ConsumerAssignException(errorMsg);
case NotAllowedError: