fix: eliminate blocking .get() in HostConnectionPool.initAsync to prevent Netty I/O thread deadlock

driver-core/src/main/java/com/datastax/driver/core/Connection.java

@@ -1328,6 +1328,44 @@ Connection open(HostConnectionPool pool, int shardId, int serverPort)
       }
     }
 
+    /**
+     * Same as {@link #open(HostConnectionPool)}, but returns a future instead of blocking.
+     *
+     * <p>This avoids blocking the calling thread (e.g. a Netty I/O thread) while waiting for the
+     * protocol handshake to complete, preventing potential cyclic thread deadlocks.
+     */
+    ListenableFuture<Connection> openAsync(final HostConnectionPool pool) {
+      return openAsync(pool, -1, 0);
+    }
+
+    /**
+     * Same as {@link #open(HostConnectionPool, int, int)}, but returns a future instead of
+     * blocking.
+     *
+     * <p>This avoids blocking the calling thread (e.g. a Netty I/O thread) while waiting for the
+     * protocol handshake to complete, preventing potential cyclic thread deadlocks.
+     */
+    ListenableFuture<Connection> openAsync(
+        final HostConnectionPool pool, final int shardId, final int serverPort) {
+      if (isShutdown)
+        return Futures.immediateFailedFuture(
+            new ConnectionException(pool.host.getEndPoint(), "Connection factory is shut down"));
+
+      pool.host.convictionPolicy.signalConnectionsOpening(1);
+      final Connection connection =
+          new Connection(buildConnectionName(pool.host), pool.host.getEndPoint(), this, pool);
+
+      return Futures.transformAsync(
+          connection.initAsync(shardId, serverPort),
+          new AsyncFunction<Void, Connection>() {
+            @Override
+            public ListenableFuture<Connection> apply(Void input) {
+              return Futures.immediateFuture(connection);
+            }
+          },
+          MoreExecutors.directExecutor());
+    }
+
     /**
      * Creates new connections and associate them to the provided connection pool, but does not
      * start them.

driver-core/src/main/java/com/datastax/driver/core/HostConnectionPool.java

@@ -276,14 +276,39 @@ ListenableFuture<Void> initAsync(Connection reusedConnection) {
     if (reusedConnection != null && reusedConnection.setOwner(this)) {
       return initAsyncWithConnection(reusedConnection);
     }
-    try {
-      return initAsyncWithConnection(manager.connectionFactory().open(this));
-    } catch (Exception e) {
-      phase.compareAndSet(Phase.INITIALIZING, Phase.INIT_FAILED);
-      SettableFuture<Void> future = SettableFuture.create();
-      future.setException(e);
-      return future;
-    }
+
+    // Use the non-blocking openAsync to avoid a cyclic deadlock. The blocking open() call could
+    // be executed on a Netty I/O thread (via directExecutor() in SessionManager.initAsync), and
+    // if Netty's round-robin assigns the new channel to the same blocked I/O thread, the driver
+    // hangs indefinitely since no timeout task can interrupt it.
+    ListenableFuture<Void> initFuture =
+        Futures.transformAsync(
+            manager.connectionFactory().openAsync(this),
+            new AsyncFunction<Connection, Void>() {
+              @Override
+              public ListenableFuture<Void> apply(Connection conn) {
+                return initAsyncWithConnection(conn);
+              }
+            },
+            MoreExecutors.directExecutor());
+
+    Futures.addCallback(
+        initFuture,
+        new FutureCallback<Void>() {
+          @Override
+          public void onSuccess(Void v) {}
+
+          @Override
+          public void onFailure(Throwable t) {
+            // initAsyncWithConnection already sets INIT_FAILED in its own callback for async
+            // connection failures; this covers the case where openAsync itself fails (e.g.
+            // factory shutdown, or immediate ECONNREFUSED before the Netty channel is created).
+            phase.compareAndSet(Phase.INITIALIZING, Phase.INIT_FAILED);
+          }
+        },
+        MoreExecutors.directExecutor());
+
+    return initFuture;
   }
 
   ListenableFuture<Void> initAsyncWithConnection(Connection reusedConnection) {

driver-core/src/test/java/com/datastax/driver/core/HostConnectionPoolTest.java

@@ -61,13 +61,15 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.common.util.concurrent.Uninterruptibles;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -320,6 +322,70 @@ public void should_close_uninitialized_pool_without_npe() throws Exception {
     }
   }
 
+  /**
+   * Ensures that {@link HostConnectionPool#initAsync(Connection)} does not block the calling thread
+   * while establishing the first connection for a new pool.
+   *
+   * <p>Previously, {@code initAsync} called the synchronous {@code Connection.Factory.open()},
+   * which internally performed an unbounded {@code .get()} on the connection handshake future. This
+   * could cause a permanent cyclic deadlock when the call was made on a Netty I/O thread (via
+   * {@code directExecutor()} chaining in {@code SessionManager.initAsync()}): if Netty's
+   * round-robin assigned the new channel to the same blocked I/O thread, neither the connection nor
+   * any timeout task queued on that thread could ever complete.
+   *
+   * <p>The fix replaces the blocking {@code open()} with the non-blocking {@code openAsync()}, so
+   * the calling thread is never stalled waiting for the protocol handshake.
+   *
+   * @test_category connection:connection_pool
+   * @see <a href="https://github.com/scylladb/scylla-dtest/pull/7046">scylla-dtest#7046</a>
+   */
+  @Test(groups = "short")
+  public void initAsync_should_not_block_calling_thread() throws Exception {
+    Cluster cluster = createClusterBuilder().build();
+    try {
+      Session session = cluster.connect();
+      Host host = TestUtils.findHost(cluster, 1);
+      SessionManager sessionManager = (SessionManager) session;
+
+      // Spy on the factory and stub openAsync() to return a future that never completes,
+      // simulating a connection attempt to a node that does not respond (e.g. firewalled).
+      Connection.Factory factory = spy(cluster.manager.connectionFactory);
+      cluster.manager.connectionFactory = factory;
+      final SettableFuture<Connection> neverCompletingFuture = SettableFuture.create();
+      doReturn(neverCompletingFuture).when(factory).openAsync(any(HostConnectionPool.class));
+
+      final HostConnectionPool pool =
+          new HostConnectionPool(host, HostDistance.LOCAL, sessionManager);
+
+      // Submit initAsync() to the actual Netty I/O thread, reproducing the exact thread context
+      // of the original bug: SessionManager.initAsync() chains updateCreatedPools() via
+      // directExecutor(), which runs on whatever thread completes the last pool-creation future —
+      // potentially a Netty I/O thread.
+      ExecutorService ioThread =
+          (ExecutorService) cluster.manager.connectionFactory.eventLoopGroup.next();
+      java.util.concurrent.Future<ListenableFuture<Void>> submitted =
+          ioThread.submit(
+              new Callable<ListenableFuture<Void>>() {
+                @Override
+                public ListenableFuture<Void> call() {
+                  return pool.initAsync(null);
+                }
+              });
+
+      // initAsync() must return a future promptly without blocking the I/O thread.
+      // With the old blocking open().get() this would time out here — and then deadlock
+      // permanently because the timeout task itself is queued on the same blocked thread.
+      ListenableFuture<Void> initFuture = submitted.get(2, TimeUnit.SECONDS);
+
+      // The returned future should still be pending because openAsync() never completed.
+      assertThat(initFuture.isDone()).isFalse();
+
+      pool.closeAsync().force();
+    } finally {
+      cluster.close();
+    }
+  }
+
   /**
    * Validates that if the keyspace tied to the Session's pool state is different than the keyspace
    * on the connection being used in dequeue that {@link Connection#setKeyspaceAsync(String)} is set