[improve][broker] Avoid logging errors when there is a connection iss…

…ue during subscription. (apache#23939) (cherry picked from commit 5e5d514)
datastax · Feb 7, 2025 · 35c97fc · 35c97fc
1 parent 560d120
commit 35c97fc
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 2 deletions.
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
@@ -224,6 +224,12 @@ public TopicPoliciesCacheNotInitException() {
         }
     }
 
+    public static class ConnectionClosedException extends BrokerServiceException {
+        public ConnectionClosedException(String msg) {
+            super(msg);
+        }
+    }
+
     public static class TopicBacklogQuotaExceededException extends BrokerServiceException {
         @Getter
         private final BacklogQuota.RetentionPolicy retentionPolicy;

diff --git a/...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/...oker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -352,7 +352,8 @@ private CompletableFuture<Consumer> internalSubscribe(final TransportCnx cnx, St
                                 consumer.consumerName(), currentUsageCount());
                     }
                     future.completeExceptionally(
-                            new BrokerServiceException("Connection was closed while the opening the cursor "));
+                            new BrokerServiceException.ConnectionClosedException(
+                                    "Connection was closed while the opening the cursor "));
                 } else {
                     log.info("[{}][{}] Created new subscription for {}", topic, subscriptionName, consumerId);
                     future.complete(consumer);

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -1004,7 +1004,8 @@ private CompletableFuture<Consumer> internalSubscribe(final TransportCnx cnx, St
 
                         decrementUsageCount();
                         return FutureUtil.failedFuture(
-                                new BrokerServiceException("Connection was closed while the opening the cursor "));
+                                new BrokerServiceException.ConnectionClosedException(
+                                        "Connection was closed while the opening the cursor "));
                     } else {
                         checkReplicatedSubscriptionControllerState();
                         if (log.isDebugEnabled()) {
@@ -1041,6 +1042,8 @@ && isCompactionSubscription(subscriptionName)) {
                     log.warn("[{}][{}] has been fenced. closing the topic {}", topic, subscriptionName,
                             ex.getMessage());
                     close();
+                } else if (ex.getCause() instanceof BrokerServiceException.ConnectionClosedException) {
+                    log.warn("[{}][{}] Connection was closed while the opening the cursor", topic, subscriptionName);
                 } else {
                     log.error("[{}] Failed to create subscription: {}", topic, subscriptionName, ex);
                 }