Use NettyFutureUtil to manage future

apache · Jan 26, 2023 · 684a135 · 684a135
1 parent dd40ff9
commit 684a135
Showing 1 changed file with 5 additions and 8 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarChannelInitializer.java
@@ -43,6 +43,7 @@
 import org.apache.pulsar.common.protocol.Commands;
 import org.apache.pulsar.common.util.SecurityUtility;
 import org.apache.pulsar.common.util.keystoretls.NettySSLContextAutoRefreshBuilder;
+import org.apache.pulsar.common.util.netty.NettyFutureUtil;
 
 @Slf4j
 public class PulsarChannelInitializer extends ChannelInitializer<SocketChannel> {
@@ -213,13 +214,11 @@ CompletableFuture<Channel> initSocks5IfConfig(Channel ch) {
     CompletableFuture<Channel> initializeClientCnx(Channel ch,
                                                    InetSocketAddress logicalAddress,
                                                    InetSocketAddress resolvedPhysicalAddress) {
-        CompletableFuture<Channel> initFuture = new CompletableFuture<>();
-        ch.eventLoop().execute(() -> {
+        return NettyFutureUtil.toCompletableFuture(ch.eventLoop().submit(() -> {
             final ClientCnx cnx = (ClientCnx) ch.pipeline().get("handler");
 
             if (cnx == null) {
-                initFuture
-                        .completeExceptionally(new IllegalStateException("Missing ClientCnx. This should not happen."));
+                throw new IllegalStateException("Missing ClientCnx. This should not happen.");
             }
 
             // Need to do our own equality because the physical address is resolved already
@@ -232,10 +231,8 @@ CompletableFuture<Channel> initializeClientCnx(Channel ch,
 
             cnx.setRemoteHostName(resolvedPhysicalAddress.getHostString());
 
-            initFuture.complete(ch);
-        });
-
-        return initFuture;
+            return ch;
+        }));
     }
 }