Merge branch '4.11.x' into 4.x

Author: adutra

changelog/README.md

@@ -7,8 +7,9 @@
 - [improvement] JAVA-2935: Make GetEntity and SetEntity methods resilient to incomplete data
 - [improvement] JAVA-2944: Upgrade MicroProfile Metrics to 3.0
 
-### 4.11.2 (in progress)
+### 4.11.2
 
+- [bug] JAVA-2932: Make DefaultDriverConfigLoader.close() resilient to terminated executors
 - [bug] JAVA-2945: Reinstate InternalDriverContext.getNodeFilter method
 - [bug] JAVA-2947: Release buffer after decoding multi-slice frame
 - [bug] JAVA-2946: Make MapperResultProducerService instances be located with user-provided class loader

core/src/main/java/com/datastax/oss/driver/api/core/config/DriverConfigLoader.java

@@ -376,7 +376,7 @@ static DriverConfigLoader compose(
   boolean supportsReloading();
 
   /**
-   * Called when the cluster closes. This is a good time to release any external resource, for
+   * Called when the session closes. This is a good time to release any external resource, for
    * example cancel a scheduled reloading task.
    */
   @Override

core/src/main/java/com/datastax/oss/driver/internal/core/config/typesafe/DefaultDriverConfigLoader.java

@@ -39,6 +39,7 @@
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import net.jcip.annotations.ThreadSafe;
@@ -235,8 +236,14 @@ public Supplier<Config> getConfigSupplier() {
   @Override
   public void close() {
     SingleThreaded singleThreaded = this.singleThreaded;
-    if (singleThreaded != null) {
-      RunOrSchedule.on(singleThreaded.adminExecutor, singleThreaded::close);
+    if (singleThreaded != null && !singleThreaded.adminExecutor.terminationFuture().isDone()) {
+      try {
+        RunOrSchedule.on(singleThreaded.adminExecutor, singleThreaded::close);
+      } catch (RejectedExecutionException e) {
+        // Checking the future is racy, there is still a tiny window that could get us here.
+        // We can safely ignore this error because, if the execution is rejected, the periodic
+        // reload task, if any, has been already cancelled.
+      }
     }
   }
 

core/src/test/java/com/datastax/oss/driver/internal/core/util/concurrent/ScheduledTaskCapturingEventLoop.java

@@ -47,7 +47,7 @@
 @SuppressWarnings("FunctionalInterfaceClash") // does not matter for test code
 public class ScheduledTaskCapturingEventLoop extends DefaultEventLoop {
 
-  private final BlockingQueue<CapturedTask> capturedTasks = new ArrayBlockingQueue<>(100);
+  private final BlockingQueue<CapturedTask<?>> capturedTasks = new ArrayBlockingQueue<>(100);
 
   public ScheduledTaskCapturingEventLoop(EventLoopGroup parent) {
     super(parent);

integration-tests/src/test/java/com/datastax/oss/driver/mapper/SelectCustomWhereClauseIT.java

@@ -16,6 +16,7 @@
 package com.datastax.oss.driver.mapper;
 
 import static com.datastax.oss.driver.assertions.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
 import static org.junit.Assume.assumeFalse;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -36,8 +37,8 @@
 import com.datastax.oss.driver.api.testinfra.session.SessionRule;
 import com.datastax.oss.driver.categories.ParallelizableTests;
 import com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures;
+import java.time.Duration;
 import java.util.concurrent.CompletionStage;
-import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
@@ -73,29 +74,35 @@ public static void setup() {
     InventoryMapper inventoryMapper =
         new SelectCustomWhereClauseIT_InventoryMapperBuilder(session).build();
     dao = inventoryMapper.productDao(SESSION_RULE.keyspace());
-  }
-
-  @Before
-  public void insertData() {
     dao.save(FLAMETHROWER);
     dao.save(MP3_DOWNLOAD);
   }
 
   @Test
   public void should_select_with_custom_clause() {
-    PagingIterable<Product> products = dao.findByDescription("%mp3%");
-    assertThat(products.one()).isEqualTo(MP3_DOWNLOAD);
-    assertThat(products.iterator()).isExhausted();
+    await()
+        .atMost(Duration.ofMinutes(1))
+        .untilAsserted(
+            () -> {
+              PagingIterable<Product> products = dao.findByDescription("%mp3%");
+              assertThat(products.one()).isEqualTo(MP3_DOWNLOAD);
+              assertThat(products.iterator()).isExhausted();
+            });
   }
 
   @Test
   public void should_select_with_custom_clause_asynchronously() {
-    MappedAsyncPagingIterable<Product> iterable =
-        CompletableFutures.getUninterruptibly(
-            dao.findByDescriptionAsync("%mp3%").toCompletableFuture());
-    assertThat(iterable.one()).isEqualTo(MP3_DOWNLOAD);
-    assertThat(iterable.currentPage().iterator()).isExhausted();
-    assertThat(iterable.hasMorePages()).isFalse();
+    await()
+        .atMost(Duration.ofMinutes(1))
+        .untilAsserted(
+            () -> {
+              MappedAsyncPagingIterable<Product> iterable =
+                  CompletableFutures.getUninterruptibly(
+                      dao.findByDescriptionAsync("%mp3%").toCompletableFuture());
+              assertThat(iterable.one()).isEqualTo(MP3_DOWNLOAD);
+              assertThat(iterable.currentPage().iterator()).isExhausted();
+              assertThat(iterable.hasMorePages()).isFalse();
+            });
   }
 
   @Mapper