From 250cb9d85a8b40feab9419cc54261ade60097d61 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Wed, 17 Jun 2026 16:13:43 +0200 Subject: [PATCH 1/2] Fix NPE in ChannelPool metrics methods when pool is uninitialized MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the initial TCP connection to a node fails, ChannelPool.connectFuture completes (so the pool is added to PoolManager.pools) while `channels` is still null — initialize() is only called on the success path. The four methods size(), getAvailableIds(), getInFlight(), and getOrphanedIds() called Arrays.stream(channels) without guarding against this, throwing NullPointerException whenever a Dropwizard Metrics reporter (JMX, Graphite, etc.) scraped the gauge values during the reconnection window. Fix: mirror the pattern already used by next(), which checks singleThreaded.initialized before touching channels. Using the volatile initialized flag is correct by the JMM: the volatile write at line 365 (initialized = true) happens-after all channels writes, so any thread that reads initialized == true is guaranteed to see channels as fully populated. Fixes: CUSTOMER-413 --- .../internal/core/pool/ChannelPool.java | 12 ++++++++ .../core/pool/ChannelPoolInitTest.java | 28 +++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java b/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java index c2aea1bb01e..8c2d2063b73 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java @@ -210,11 +210,17 @@ public DriverChannel next(@Nullable Token routingKey, @Nullable Integer shardSug /** @return the number of active channels in the pool. */ public int size() { + if (!singleThreaded.initialized) { + return 0; + } return Arrays.stream(channels).mapToInt(ChannelSet::size).sum(); } /** @return the number of available stream ids on all channels in the pool. */ public int getAvailableIds() { + if (!singleThreaded.initialized) { + return 0; + } return Arrays.stream(channels).mapToInt(ChannelSet::getAvailableIds).sum(); } @@ -223,6 +229,9 @@ public int getAvailableIds() { * {@link #getOrphanedIds() orphaned ids}). */ public int getInFlight() { + if (!singleThreaded.initialized) { + return 0; + } return Arrays.stream(channels).mapToInt(ChannelSet::getInFlight).sum(); } @@ -232,6 +241,9 @@ public int getInFlight() { * might still come from the server. */ public int getOrphanedIds() { + if (!singleThreaded.initialized) { + return 0; + } return Arrays.stream(channels).mapToInt(ChannelSet::getOrphanedIds).sum(); } diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolInitTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolInitTest.java index aa3eab0e1bb..d4fb0211173 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolInitTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolInitTest.java @@ -148,6 +148,34 @@ public void should_fire_force_down_event_when_cluster_name_does_not_match() thro factoryHelper.verifyNoMoreCalls(); } + @Test + public void should_return_zero_for_metric_accessors_when_pool_uninitialized() throws Exception { + // Reproduce CUSTOMER-413: when the initial connection attempt fails, connectFuture completes + // with channels == null (initialize() is only called on the success path). Before the fix, + // any call to size/getAvailableIds/getInFlight/getOrphanedIds threw NullPointerException via + // Arrays.stream(null), which propagated through the Dropwizard Metrics gauge lambdas + // registered in DropwizardNodeMetricUpdater. + when(defaultProfile.getInt(DefaultDriverOption.CONNECTION_POOL_LOCAL_SIZE)).thenReturn(1); + + MockChannelFactoryHelper.builder(channelFactory) + .failure(node, "mock channel init failure") + .build(); + + CompletionStage poolFuture = + ChannelPool.init(node, null, NodeDistance.LOCAL, context, "test"); + + ChannelPool pool = poolFuture.toCompletableFuture().get(); + + // Confirm the precondition: pool entered the map but channels was never initialized + assertThat(pool.channels).isNull(); + + // All four accessor methods must return 0 without throwing + assertThat(pool.size()).isEqualTo(0); + assertThat(pool.getAvailableIds()).isEqualTo(0); + assertThat(pool.getInFlight()).isEqualTo(0); + assertThat(pool.getOrphanedIds()).isEqualTo(0); + } + @Test public void should_reconnect_when_init_incomplete() throws Exception { // Short delay so we don't have to wait in the test From fdcbf39024a94f68de2ca4228c1b4aa22691e919 Mon Sep 17 00:00:00 2001 From: Dmitry Kropachev Date: Tue, 23 Jun 2026 13:33:04 -0400 Subject: [PATCH 2/2] Fix closing uninitialized channel pools If the initial pool connection attempt fails, ChannelPool completes while still uninitialized and starts a reconnect loop. A session close during that state can complete immediately because there are no active channels yet. When the pending reconnect later succeeds, the uninitialized reconnect path previously initialized the pool and kept the new channel open even though the pool had already been closed. This could leave a leaked channel and incorrectly publish channelOpened after shutdown. Guard that reconnect-success path with isClosing: force-close the new channel and complete the reconnect attempt without initializing the pool. Add coverage for both closeAsync and forceCloseAsync when the first successful channel arrives after shutdown. --- .../internal/core/pool/ChannelPool.java | 7 ++ .../core/pool/ChannelPoolShutdownTest.java | 68 +++++++++++++++++++ 2 files changed, 75 insertions(+) diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java b/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java index 8c2d2063b73..c9bc5df2f85 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/pool/ChannelPool.java @@ -418,6 +418,13 @@ private CompletionStage 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); diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolShutdownTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolShutdownTest.java index b40bcb4aa39..8bc08e4c6c8 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/pool/ChannelPoolShutdownTest.java +++ b/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 channel1Future = new CompletableFuture<>(); + MockChannelFactoryHelper factoryHelper = + MockChannelFactoryHelper.builder(channelFactory) + // init + .failure(node, "mock channel init failure") + // reconnection + .pending(node, channel1Future) + .build(); + + CompletionStage poolFuture = + ChannelPool.init(node, null, NodeDistance.LOCAL, context, "test"); + + factoryHelper.waitForCalls(node, 2); + assertThatStage(poolFuture).isSuccess(); + ChannelPool pool = poolFuture.toCompletableFuture().get(); + + CompletionStage 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 channel1Future = new CompletableFuture<>(); + MockChannelFactoryHelper factoryHelper = + MockChannelFactoryHelper.builder(channelFactory) + // init + .failure(node, "mock channel init failure") + // reconnection + .pending(node, channel1Future) + .build(); + + CompletionStage poolFuture = + ChannelPool.init(node, null, NodeDistance.LOCAL, context, "test"); + + factoryHelper.waitForCalls(node, 2); + assertThatStage(poolFuture).isSuccess(); + ChannelPool pool = poolFuture.toCompletableFuture().get(); + + CompletionStage closeFuture = pool.forceCloseAsync(); + assertThatStage(closeFuture).isSuccess(); + + channel1Future.complete(channel1); + + verify(channel1, VERIFY_TIMEOUT).forceClose(); + verify(eventBus, never()).fire(ChannelEvent.channelOpened(node)); + + factoryHelper.verifyNoMoreCalls(); + } }