GEODE-9040: Shutdown ExecutorService when the SingleThreadColocationLogger is stopped

geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/colocation/SingleThreadColocationLogger.java

@@ -63,7 +63,7 @@ public class SingleThreadColocationLogger implements ColocationLogger {
     this(region, delayMillis, intervalMillis, LOGGER::warn,
         pr -> getAllColocationRegions(pr).keySet(),
         newSingleThreadExecutor(
-            runnable -> new LoggingThread("ColocationLogger for " + region.getName(), false,
+            runnable -> new LoggingThread("ColocationLogger for " + region.getName(), true,
                 runnable)));
   }
 
@@ -95,6 +95,7 @@ public ColocationLogger start() {
   public void stop() {
     synchronized (lock) {
       missingChildren.clear();
+      executorService.shutdownNow();
       lock.notifyAll();
     }
   }
@@ -151,6 +152,11 @@ List<String> getMissingChildren() {
     return new ArrayList<>(missingChildren);
   }
 
+  @VisibleForTesting
+  ExecutorService getExecutorService() {
+    return executorService;
+  }
+
   private Runnable checkForMissingColocatedRegionRunnable() {
     return this::checkForMissingColocatedRegion;
   }

geode-core/src/test/java/org/apache/geode/internal/cache/partitioned/colocation/SingleThreadColocationLoggerTest.java

@@ -17,6 +17,7 @@
 import static java.lang.System.lineSeparator;
 import static java.util.Collections.singleton;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.catchThrowable;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -213,4 +214,18 @@ public void updateAndGetMissingChildRegionsUpdatesMissingChildren() {
     assertThat(colocationLogger.getMissingChildren())
         .isEmpty();
   }
+
+  @Test
+  public void stopTerminatesExecutorService() {
+    SingleThreadColocationLogger colocationLogger =
+        new SingleThreadColocationLogger(region, 500, 1000, logger,
+            allColocationRegionsProvider, executorService);
+    colocationLogger.start();
+
+    colocationLogger.stop();
+
+    // Wait until the ExecutorService is terminated
+    await().untilAsserted(
+        () -> assertThat(colocationLogger.getExecutorService().isTerminated()).isTrue());
+  }
 }