Tolerate DefaultConnectionPool.ready/invalidate being called after or concurrently with close (#906)

JAVA-4551

Author: stIncMale

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -69,6 +69,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -1416,18 +1417,21 @@ private BackgroundMaintenanceManager() {
         void start() {
             if (maintainer != null) {
                 assertTrue(cancellationHandle == null);
-                cancellationHandle = maintainer.scheduleAtFixedRate(DefaultConnectionPool.this::doMaintenance,
+                cancellationHandle = ignoreRejectedExectution(() -> maintainer.scheduleAtFixedRate(
+                        DefaultConnectionPool.this::doMaintenance,
                         initialStart ? settings.getMaintenanceInitialDelay(MILLISECONDS) : 0,
-                        settings.getMaintenanceFrequency(MILLISECONDS), MILLISECONDS);
+                        settings.getMaintenanceFrequency(MILLISECONDS), MILLISECONDS));
                 initialStart = false;
             }
         }
 
         void runOnceAndStop() {
             if (maintainer != null) {
-                assertNotNull(cancellationHandle).cancel(false);
-                cancellationHandle = null;
-                maintainer.execute(DefaultConnectionPool.this::doMaintenance);
+                if (cancellationHandle != null) {
+                    cancellationHandle.cancel(false);
+                    cancellationHandle = null;
+                }
+                ignoreRejectedExectution(() -> maintainer.execute(DefaultConnectionPool.this::doMaintenance));
             }
         }
 
@@ -1437,6 +1441,23 @@ public void close() {
                 maintainer.shutdownNow();
             }
         }
+
+        private void ignoreRejectedExectution(final Runnable action) {
+            ignoreRejectedExectution(() -> {
+                action.run();
+                return null;
+            });
+        }
+
+        @Nullable
+        private <T> T ignoreRejectedExectution(final Supplier<T> action) {
+            try {
+                return action.get();
+            } catch (RejectedExecutionException ignored) {
+                // `close` either completed or is in progress
+                return null;
+            }
+        }
     }
 
     @ThreadSafe

driver-core/src/test/functional/com/mongodb/internal/connection/DefaultConnectionPoolTest.java

@@ -443,6 +443,52 @@ public void checkoutHandOverMechanism() throws InterruptedException, TimeoutExce
         }
     }
 
+    @Test
+    public void readyAfterCloseMustNotThrow() {
+        provider = new DefaultConnectionPool(
+                SERVER_ID,
+                connectionFactory,
+                ConnectionPoolSettings.builder().maxSize(1).build(),
+                mockSdamProvider());
+        provider.close();
+        provider.ready();
+    }
+
+    @Test
+    public void invalidateAfterCloseMustNotThrow() {
+        provider = new DefaultConnectionPool(
+                SERVER_ID,
+                connectionFactory,
+                ConnectionPoolSettings.builder().maxSize(1).build(),
+                mockSdamProvider());
+        provider.ready();
+        provider.close();
+        provider.invalidate(null);
+    }
+
+    @Test
+    public void readyInvalidateConcurrentWithCloseMustNotThrow() throws ExecutionException, InterruptedException {
+        Future<?> readyAndInvalidateResult = null;
+        for (int i = 0; i < 3_000; i++) {
+            provider = new DefaultConnectionPool(
+                    SERVER_ID,
+                    connectionFactory,
+                    ConnectionPoolSettings.builder().maxSize(1).build(),
+                    mockSdamProvider());
+            try {
+                readyAndInvalidateResult = cachedExecutor.submit(() -> {
+                    provider.ready();
+                    provider.invalidate(null);
+                });
+            } finally {
+                provider.close();
+                if (readyAndInvalidateResult != null) {
+                    readyAndInvalidateResult.get();
+                }
+            }
+        }
+    }
+
     private static void assertUseConcurrently(final DefaultConnectionPool pool, final int concurrentUsersCount,
                                               final boolean sync, final boolean async,
                                               final float invalidateAndReadyProb, final float invalidateProb, final float readyProb,