[improve][proxy] Only create ConnectionPool when needed (apache#20062)

(cherry picked from commit 36579dd)
michaeljmarshall · Apr 19, 2023 · 6332aa4 · 6332aa4
1 parent f837092
commit 6332aa4
Showing 1 changed file with 23 additions and 22 deletions.
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConnection.java
@@ -75,7 +75,7 @@
  */
 public class ProxyConnection extends PulsarHandler {
     private static final Logger LOG = LoggerFactory.getLogger(ProxyConnection.class);
-    // ConnectionPool is used by the proxy to issue lookup requests
+    // ConnectionPool is used by the proxy to issue lookup requests. It is null when doing direct broker proxying.
     private ConnectionPool connectionPool;
     private final AtomicLong requestIdGenerator =
             new AtomicLong(ThreadLocalRandom.current().nextLong(0, Long.MAX_VALUE / 2));
@@ -260,24 +260,7 @@ public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exce
     }
 
     private synchronized void completeConnect() throws PulsarClientException {
-        Supplier<ClientCnx> clientCnxSupplier;
-        if (service.getConfiguration().isAuthenticationEnabled()) {
-            clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole,
-                    clientAuthData, clientAuthMethod, protocolVersionToAdvertise,
-                    service.getConfiguration().isForwardAuthorizationCredentials(), this);
-        } else {
-            clientCnxSupplier = () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise);
-        }
-
-        if (this.connectionPool == null) {
-            this.connectionPool = new ConnectionPool(clientConf, service.getWorkerGroup(),
-                    clientCnxSupplier,
-                    Optional.of(dnsAddressResolverGroup.getResolver(service.getWorkerGroup().next())));
-        } else {
-            LOG.error("BUG! Connection Pool has already been created for proxy connection to {} state {} role {}",
-                    remoteAddress, state, clientAuthRole);
-        }
-
+        checkArgument(state == State.Connecting);
         LOG.info("[{}] complete connection, init proxy handler. authenticated with {} role {}, hasProxyToBrokerUrl: {}",
                 remoteAddress, authMethod, clientAuthRole, hasProxyToBrokerUrl);
         if (hasProxyToBrokerUrl) {
@@ -321,8 +304,26 @@ private synchronized void completeConnect() throws PulsarClientException {
                     });
         } else {
             // Client is doing a lookup, we can consider the handshake complete
-            // and we'll take care of just topics and
-            // partitions metadata lookups
+            // and we'll take care of just topics and partitions metadata lookups
+            Supplier<ClientCnx> clientCnxSupplier;
+            if (service.getConfiguration().isAuthenticationEnabled()) {
+                clientCnxSupplier = () -> new ProxyClientCnx(clientConf, service.getWorkerGroup(), clientAuthRole,
+                        clientAuthData, clientAuthMethod, protocolVersionToAdvertise,
+                        service.getConfiguration().isForwardAuthorizationCredentials(), this);
+            } else {
+                clientCnxSupplier =
+                        () -> new ClientCnx(clientConf, service.getWorkerGroup(), protocolVersionToAdvertise);
+            }
+
+            if (this.connectionPool == null) {
+                this.connectionPool = new ConnectionPool(clientConf, service.getWorkerGroup(),
+                        clientCnxSupplier,
+                        Optional.of(dnsAddressResolverGroup.getResolver(service.getWorkerGroup().next())));
+            } else {
+                LOG.error("BUG! Connection Pool has already been created for proxy connection to {} state {} role {}",
+                        remoteAddress, state, clientAuthRole);
+            }
+
             state = State.ProxyLookupRequests;
             try {
                 lookupProxyHandler =
@@ -409,7 +410,7 @@ protected void authChallengeSuccessCallback(AuthData authChallenge) {
                 }
 
                 // First connection
-                if (this.connectionPool == null || state == State.Connecting) {
+                if (state == State.Connecting) {
                     // authentication has completed, will send newConnected command.
                     completeConnect();
                 }