Internal Producer/Consumer in WebSocketProxy should be closed asynchronously (#473)

pulsar-websocket/src/main/java/com/yahoo/pulsar/websocket/ConsumerHandler.java

@@ -135,7 +135,7 @@ public class ConsumerHandler extends AbstractWebSocketHandler {
                             @Override
                             public void writeSuccess() {
                                 if (log.isDebugEnabled()) {
-                                    log.info("[{}/{}] message is delivered successfully to {} ", consumer.getTopic(),
+                                    log.debug("[{}/{}] message is delivered successfully to {} ", consumer.getTopic(),
                                             subscription, getRemote().getInetSocketAddress().toString());
                                 }
                                 updateDeliverMsgStat(msgSize);
@@ -184,7 +184,14 @@ public class ConsumerHandler extends AbstractWebSocketHandler {
     public void close() throws IOException {
         if (consumer != null) {
             this.service.removeConsumer(this);
-            consumer.close();
+            consumer.closeAsync().thenAccept(x -> {
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Closed consumer asynchronously", consumer.getTopic());
+                }
+            }).exceptionally(exception -> {
+                log.warn("[{}] Failed to close consumer", consumer.getTopic(), exception);
+                return null;
+            });
         }
     }
 

pulsar-websocket/src/main/java/com/yahoo/pulsar/websocket/ProducerHandler.java

@@ -83,7 +83,14 @@ public class ProducerHandler extends AbstractWebSocketHandler {
     public void close() throws IOException {
         if (producer != null) {
             this.service.removeProducer(this);
-            producer.close();
+            producer.closeAsync().thenAccept(x -> {
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Closed producer asynchronously", producer.getTopic());
+                }
+            }).exceptionally(exception -> {
+                log.warn("[{}] Failed to close producer", producer.getTopic(), exception);
+                return null;
+            });
         }
     }