[client-v2] Do not close provided executor

client-v2/src/main/java/com/clickhouse/client/api/Client.java

@@ -145,6 +145,8 @@ public class Client implements AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(Client.class);
     private final ExecutorService sharedOperationExecutor;
 
+    private final boolean isSharedOpExecuterorOwned;
+
     private final Map<String, ClientStatisticsHolder> globalClientStats = new ConcurrentHashMap<>();
 
     private boolean useNewImplementation = false;
@@ -172,8 +174,10 @@ private Client(Set<String> endpoints, Map<String,String> configuration, boolean
 
         boolean isAsyncEnabled = MapUtils.getFlag(this.configuration, ClientConfigProperties.ASYNC_OPERATIONS.getKey(), false);
         if (isAsyncEnabled && sharedOperationExecutor == null) {
+            this.isSharedOpExecuterorOwned = true;
             this.sharedOperationExecutor = Executors.newCachedThreadPool(new DefaultThreadFactory("chc-operation"));
         } else {
+            this.isSharedOpExecuterorOwned = false;
             this.sharedOperationExecutor = sharedOperationExecutor;
         }
         this.useNewImplementation = useNewImplementation;
@@ -225,12 +229,16 @@ public String getDefaultDatabase() {
      */
     @Override
     public void close() {
-        try {
-            if (sharedOperationExecutor != null && !sharedOperationExecutor.isShutdown()) {
-                this.sharedOperationExecutor.shutdownNow();
+        if (isSharedOpExecuterorOwned) {
+            try {
+                if (sharedOperationExecutor != null && !sharedOperationExecutor.isShutdown()) {
+                    this.sharedOperationExecutor.shutdownNow();
+                }
+            } catch (Exception e) {
+                LOG.error("Failed to close shared operation executor", e);
             }
-        } catch (Exception e) {
-            LOG.error("Failed to close shared operation executor", e);
+        } else {
+            LOG.debug("Skip closing operation executor because not owned by client");
         }
 
         if (oldClient != null) {
@@ -777,9 +785,10 @@ public Builder useAsyncRequests(boolean async) {
         /**
          * Sets an executor for running operations. If async operations are enabled and no executor is specified
          * client will create a default executor.
-         *
+         * Executor will stay running after {@code Client#close() } is called. It is application responsibility to close
+         * the executor.
          * @param executorService - executor service for async operations
-         * @return
+         * @return 
          */
         public Builder setSharedOperationExecutor(ExecutorService executorService) {
             this.sharedOperationExecutor = executorService;

client-v2/src/test/java/com/clickhouse/client/ClientTests.java

@@ -5,6 +5,7 @@
 import com.clickhouse.client.api.ClientException;
 import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.query.GenericRecord;
+import com.clickhouse.client.api.query.QueryResponse;
 import com.clickhouse.client.api.query.QuerySettings;
 import com.clickhouse.client.api.query.Records;
 import com.clickhouse.client.config.ClickHouseClientOption;
@@ -19,7 +20,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class ClientTests extends BaseIntegrationTest {
     private static final Logger LOGGER = LoggerFactory.getLogger(ClientTests.class);
@@ -130,6 +134,25 @@ public void testSetOptions() {
         }
     }
 
+    @Test
+    public void testProvidedExecutor() throws Exception {
+
+        ExecutorService executorService = Executors.newSingleThreadExecutor();
+        try (Client client = newClient().useAsyncRequests(true).setSharedOperationExecutor(executorService).build()) {
+            QueryResponse response = client.query("SELECT 1").get();
+            response.getMetrics();
+        } catch (Exception e) {
+            Assert.fail("unexpected exception", e);
+        }
+
+        AtomicBoolean flag = new AtomicBoolean(true);
+        executorService.submit(() -> flag.compareAndSet(true, false));
+        executorService.shutdown();
+        executorService.awaitTermination(10, TimeUnit.SECONDS);
+
+        Assert.assertFalse(flag.get());
+    }
+
     protected Client.Builder newClient() {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
         boolean isSecure = isCloud();