GH-1277 SafeNotifyService threads leak in CuratorFrameWorkImpl

CURATOR-495 introduced a new runSafeService field in CuratorFrameworkImpl class, and this field is
either initialized by an external ExecutorService via the builder, or it is created internally
within the class.

In the CuratorFrameworkImpl#close method though, this Executor is never closed, so the threads
that are opened by the instances are lingering there until the VM is closed by default.
Worse, if someone specifies a thread factory to the framework implementation via the builder that
produces non-daemon threads, the VM never exits due to the unstopped single thread executor.

Author: fapifta

curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java

@@ -100,6 +100,7 @@ public final class CuratorFrameworkImpl extends CuratorFrameworkBase {
     private final EnsembleTracker ensembleTracker;
     private final SchemaSet schemaSet;
     private final Executor runSafeService;
+    private boolean isExternalRunSafeService = false;
     private final ZookeeperCompatibility zookeeperCompatibility;
 
     private volatile ExecutorService executorService;
@@ -194,6 +195,7 @@ public void process(WatchedEvent watchedEvent) {
 
     private Executor makeRunSafeService(CuratorFrameworkFactory.Builder builder) {
         if (builder.getRunSafeService() != null) {
+            isExternalRunSafeService = true;
             return builder.getRunSafeService();
         }
         ThreadFactory threadFactory = builder.getThreadFactory();
@@ -383,6 +385,16 @@ public void close() {
                 }
             }
 
+            if (!isExternalRunSafeService) {
+                ((ExecutorService) runSafeService).shutdownNow();
+                try {
+                    ((ExecutorService) runSafeService).awaitTermination(maxCloseWaitMs, TimeUnit.MILLISECONDS);
+                } catch (InterruptedException e) {
+                    // Interrupted while interrupting; I give up.
+                    Thread.currentThread().interrupt();
+                }
+            }
+
             if (ensembleTracker != null) {
                 ensembleTracker.close();
             }

curator-framework/src/test/java/org/apache/curator/framework/imps/TestFrameworkBackground.java

@@ -24,14 +24,19 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Queues;
+import io.netty.util.concurrent.DefaultThreadFactory;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.api.ACLProvider;
@@ -45,6 +50,7 @@
 import org.apache.curator.test.BaseClassForTests;
 import org.apache.curator.test.Timing;
 import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.utils.ThreadUtils;
 import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.data.ACL;
 import org.junit.jupiter.api.Test;
@@ -306,4 +312,49 @@ public void listen(OperationAndData<?> data) {
             CloseableUtils.closeQuietly(client);
         }
     }
+
+    @Test
+    public void testCloseShutsDownInternalRunSafeService() {
+        Timing timing = new Timing();
+        CuratorFramework client = CuratorFrameworkFactory.newClient(
+                server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+        client.start();
+        client.runSafe(() -> {});
+        assertTrue(enumerateThreads().anyMatch(t -> t.getName().contains("SafeNotifyService")));
+
+        client.close();
+
+        assertTrue(enumerateThreads().noneMatch(t -> t.getName().contains("SafeNotifyService")));
+    }
+
+    @Test
+    public void testCloseLeavesExternalRunSafeServiceRunning() throws Exception {
+        Timing timing = new Timing();
+        ExecutorService externalRunner =
+                Executors.newSingleThreadScheduledExecutor(ThreadUtils.newThreadFactory("ExternalSafeNotifyService"));
+        CuratorFramework client = CuratorFrameworkFactory.builder()
+                .connectString(server.getConnectString())
+                .sessionTimeoutMs(timing.session())
+                .connectionTimeoutMs(timing.connection())
+                .retryPolicy(new RetryOneTime(1))
+                .maxCloseWaitMs(timing.forWaiting().milliseconds())
+                .runSafeService(externalRunner)
+                .build();
+        client.start();
+        client.runSafe(() -> {});
+        assertTrue(enumerateThreads().anyMatch(t -> t.getName().contains("ExternalSafeNotifyService")));
+
+        client.close();
+
+        assertTrue(enumerateThreads().anyMatch(t -> t.getName().contains("ExternalSafeNotifyService")));
+
+        externalRunner.shutdownNow();
+        assertTrue(externalRunner.awaitTermination(10, TimeUnit.SECONDS));
+    }
+
+    private static Stream<Thread> enumerateThreads() {
+        Thread[] threads = new Thread[Thread.activeCount()];
+        Thread.enumerate(threads);
+        return Arrays.stream(threads);
+    }
 }