Guard against NPE in MongoMetricsConnectionPoolListener (#2454)

Logically, the other events should not happen after the connection pool has been closed, but it seems they sometimes do. When they do (e.g. on shutdown), a NPE was being thrown. This change guards against such NPEs.

Resolves gh-2384

micrometer-core/src/main/java/io/micrometer/core/instrument/binder/mongodb/MongoMetricsConnectionPoolListener.java

@@ -81,38 +81,50 @@ public void connectionPoolClosed(ConnectionPoolClosedEvent event) {
 
     @Override
     public void connectionCheckedOut(ConnectionCheckedOutEvent event) {
-        checkedOutCount.get(event.getConnectionId().getServerId())
-                .incrementAndGet();
+        AtomicInteger gauge = checkedOutCount.get(event.getConnectionId().getServerId());
+        if (gauge != null) {
+            gauge.incrementAndGet();
+        }
     }
 
     @Override
     public void connectionCheckedIn(ConnectionCheckedInEvent event) {
-        checkedOutCount.get(event.getConnectionId().getServerId())
-                .decrementAndGet();
+        AtomicInteger gauge = checkedOutCount.get(event.getConnectionId().getServerId());
+        if (gauge != null) {
+            gauge.decrementAndGet();
+        }
     }
 
     @Override
     public void waitQueueEntered(ConnectionPoolWaitQueueEnteredEvent event) {
-        waitQueueSize.get(event.getServerId())
-                .incrementAndGet();
+        AtomicInteger gauge = waitQueueSize.get(event.getServerId());
+        if (gauge != null) {
+            gauge.incrementAndGet();
+        }
     }
 
     @Override
     public void waitQueueExited(ConnectionPoolWaitQueueExitedEvent event) {
-        waitQueueSize.get(event.getServerId())
-                .decrementAndGet();
+        AtomicInteger gauge = waitQueueSize.get(event.getServerId());
+        if (gauge != null) {
+            gauge.decrementAndGet();
+        }
     }
 
     @Override
     public void connectionAdded(ConnectionAddedEvent event) {
-        poolSize.get(event.getConnectionId().getServerId())
-                .incrementAndGet();
+        AtomicInteger gauge = poolSize.get(event.getConnectionId().getServerId());
+        if (gauge != null) {
+            gauge.incrementAndGet();
+        }
     }
 
     @Override
     public void connectionRemoved(ConnectionRemovedEvent event) {
-        poolSize.get(event.getConnectionId().getServerId())
-                .decrementAndGet();
+        AtomicInteger gauge = poolSize.get(event.getConnectionId().getServerId());
+        if (gauge != null) {
+            gauge.decrementAndGet();
+        }
     }
 
     private Gauge registerGauge(ServerId serverId, String metricName, String description, Map<ServerId, AtomicInteger> metrics) {

micrometer-core/src/test/java/io/micrometer/core/instrument/binder/mongodb/MongoMetricsConnectionPoolListenerTest.java

@@ -18,8 +18,12 @@
 import com.mongodb.MongoClient;
 import com.mongodb.MongoClientOptions;
 import com.mongodb.ServerAddress;
-import com.mongodb.event.ClusterListenerAdapter;
-import com.mongodb.event.ClusterOpeningEvent;
+import com.mongodb.connection.ClusterId;
+import com.mongodb.connection.ConnectionId;
+import com.mongodb.connection.ConnectionPoolSettings;
+import com.mongodb.connection.ServerId;
+import com.mongodb.event.*;
+import io.micrometer.core.Issue;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Tags;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
@@ -28,6 +32,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 /**
  * Tests for {@link MongoMetricsConnectionPoolListener}.
@@ -36,9 +41,10 @@
  */
 class MongoMetricsConnectionPoolListenerTest extends AbstractMongoDbTest {
 
+    private final MeterRegistry registry = new SimpleMeterRegistry();
+
     @Test
     void shouldCreatePoolMetrics() {
-        MeterRegistry registry = new SimpleMeterRegistry();
         AtomicReference<String> clusterId = new AtomicReference<>();
         MongoClientOptions options = MongoClientOptions.builder()
                 .minConnectionsPerHost(2)
@@ -71,4 +77,17 @@ public void clusterOpening(ClusterOpeningEvent event) {
                 .isNull();
     }
 
+    @Test
+    @Issue("#2384")
+    void whenConnectionCheckedInAfterPoolClose_thenNoExceptionThrown() {
+        ServerId serverId = new ServerId(new ClusterId(), new ServerAddress(HOST, port));
+        ConnectionId connectionId = new ConnectionId(serverId);
+        MongoMetricsConnectionPoolListener listener = new MongoMetricsConnectionPoolListener(registry);
+        listener.connectionPoolOpened(new ConnectionPoolOpenedEvent(serverId, ConnectionPoolSettings.builder().build()));
+        listener.connectionCheckedOut(new ConnectionCheckedOutEvent(connectionId));
+        listener.connectionPoolClosed(new ConnectionPoolClosedEvent(serverId));
+        assertThatCode(() -> listener.connectionCheckedIn(new ConnectionCheckedInEvent(connectionId)))
+                .doesNotThrowAnyException();
+    }
+
 }