Fix closing uninitialized channel pools

Author: dkropachev

core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java

@@ -418,6 +418,13 @@ private CompletionStage<Boolean> reconnect() {
                 } else {
                   resultFuture.complete(false);
                 }
+              } else if (isClosing) {
+                LOG.debug(
+                    "[{}] New channel added ({}) but the pool was closed, closing it",
+                    logPrefix,
+                    driver);
+                driver.forceClose();
+                resultFuture.complete(true);
               } else {
                 initialize(driver);
                 CompletableFutures.completeFrom(addMissingChannels(), resultFuture);

core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolShutdownTest.java

@@ -104,6 +104,40 @@ public void should_close_all_channels_when_closed() throws Exception {
     factoryHelper.verifyNoMoreCalls();
   }
 
+  @Test
+  public void should_force_close_first_reconnected_channel_when_closed_before_initialized()
+      throws Exception {
+    when(reconnectionSchedule.nextDelay()).thenReturn(Duration.ofNanos(1));
+    when(defaultProfile.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)).thenReturn(1);
+
+    DriverChannel channel1 = newMockDriverChannel(1);
+    CompletableFuture<DriverChannel> channel1Future = new CompletableFuture<>();
+    MockChannelFactoryHelper factoryHelper =
+        MockChannelFactoryHelper.builder(channelFactory)
+            // init
+            .failure(node, "mock channel init failure")
+            // reconnection
+            .pending(node, channel1Future)
+            .build();
+
+    CompletionStage<ChannelPool> poolFuture =
+        ChannelPool.init(node, null, NodeDistance.LOCAL, context, "test");
+
+    factoryHelper.waitForCalls(node, 2);
+    assertThatStage(poolFuture).isSuccess();
+    ChannelPool pool = poolFuture.toCompletableFuture().get();
+
+    CompletionStage<Void> closeFuture = pool.closeAsync();
+    assertThatStage(closeFuture).isSuccess();
+
+    channel1Future.complete(channel1);
+
+    verify(channel1, VERIFY_TIMEOUT).forceClose();
+    verify(eventBus, never()).fire(ChannelEvent.channelOpened(node));
+
+    factoryHelper.verifyNoMoreCalls();
+  }
+
   @Test
   public void should_force_close_all_channels_when_force_closed() throws Exception {
     when(reconnectionSchedule.nextDelay()).thenReturn(Duration.ofNanos(1));
@@ -170,4 +204,38 @@ public void should_force_close_all_channels_when_force_closed() throws Exception
 
     factoryHelper.verifyNoMoreCalls();
   }
+
+  @Test
+  public void should_force_close_first_reconnected_channel_when_force_closed_before_initialized()
+      throws Exception {
+    when(reconnectionSchedule.nextDelay()).thenReturn(Duration.ofNanos(1));
+    when(defaultProfile.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)).thenReturn(1);
+
+    DriverChannel channel1 = newMockDriverChannel(1);
+    CompletableFuture<DriverChannel> channel1Future = new CompletableFuture<>();
+    MockChannelFactoryHelper factoryHelper =
+        MockChannelFactoryHelper.builder(channelFactory)
+            // init
+            .failure(node, "mock channel init failure")
+            // reconnection
+            .pending(node, channel1Future)
+            .build();
+
+    CompletionStage<ChannelPool> poolFuture =
+        ChannelPool.init(node, null, NodeDistance.LOCAL, context, "test");
+
+    factoryHelper.waitForCalls(node, 2);
+    assertThatStage(poolFuture).isSuccess();
+    ChannelPool pool = poolFuture.toCompletableFuture().get();
+
+    CompletionStage<Void> closeFuture = pool.forceCloseAsync();
+    assertThatStage(closeFuture).isSuccess();
+
+    channel1Future.complete(channel1);
+
+    verify(channel1, VERIFY_TIMEOUT).forceClose();
+    verify(eventBus, never()).fire(ChannelEvent.channelOpened(node));
+
+    factoryHelper.verifyNoMoreCalls();
+  }
 }