Log warn message if exception occurs while WebSocket proxy is sending…

…/receiving message (apache#1537)
iEverX · Apr 10, 2018 · f4300bd · f4300bd
1 parent 9525047
commit f4300bd
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 2 deletions.
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
@@ -35,6 +35,7 @@
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerConfiguration;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
 import org.apache.pulsar.client.api.PulsarClientException.ConsumerBusyException;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -181,6 +182,14 @@ public void writeSuccess() {
                 service.getExecutor().execute(() -> receiveMessage());
             }
         }).exceptionally(exception -> {
+            if (exception.getCause() instanceof AlreadyClosedException) {
+                log.info("[{}/{}] Consumer was closed while receiving msg from broker", consumer.getTopic(),
+                        subscription);
+            } else {
+                log.warn("[{}/{}] Error occurred while consumer handler was delivering msg to {}: {}",
+                        consumer.getTopic(), subscription, getRemote().getInetSocketAddress().toString(),
+                        exception.getMessage());
+            }
             return null;
         });
     }

diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
@@ -187,6 +187,8 @@ public void onWebSocketText(String message) {
                 sendAckResponse(new ProducerAck(messageId, sendRequest.context));
             }
         }).exceptionally(exception -> {
+            log.warn("[{}] Error occurred while producer handler was sending msg from {}: {}", producer.getTopic(),
+                    getRemote().getInetSocketAddress().toString(), exception.getMessage());
             numMsgsFailed.increment();
             sendAckResponse(
                     new ProducerAck(UnknownError, exception.getMessage(), null, sendRequest.context));

diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
@@ -31,6 +31,7 @@
 import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.ReaderBuilder;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -166,8 +167,12 @@ public void writeSuccess() {
                 service.getExecutor().execute(() -> receiveMessage());
             }
         }).exceptionally(exception -> {
-            log.warn("[{}/{}] Failed to deliver msg to {} {}", reader.getTopic(),
-                    subscription, getRemote().getInetSocketAddress().toString(), exception);
+            if (exception.getCause() instanceof AlreadyClosedException) {
+                log.info("[{}/{}] Reader was closed while receiving msg from broker", reader.getTopic(), subscription);
+            } else {
+                log.warn("[{}/{}] Error occurred while reader handler was delivering msg to {}: {}", reader.getTopic(),
+                        subscription, getRemote().getInetSocketAddress().toString(), exception.getMessage());
+            }
             return null;
         });
     }