fixed NPE and added test

Author: chernser

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

@@ -2153,7 +2153,11 @@ private void applyDefaults(QuerySettings settings) {
 
     private <T> CompletableFuture<T> runAsyncOperation(Supplier<T> resultSupplier, Map<String, Object> requestSettings) {
         boolean isAsync = MapUtils.getFlag(requestSettings, configuration, ClientConfigProperties.ASYNC_OPERATIONS.getKey());
-        return isAsync ? CompletableFuture.supplyAsync(resultSupplier, sharedOperationExecutor) : CompletableFuture.completedFuture(resultSupplier.get());
+        if (isAsync) {
+            return sharedOperationExecutor == null ? CompletableFuture.supplyAsync(resultSupplier) :
+                    CompletableFuture.supplyAsync(resultSupplier, sharedOperationExecutor);
+        }
+        return CompletableFuture.completedFuture(resultSupplier.get());
     }
 
     @Override

client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java

@@ -5,6 +5,7 @@
 import com.clickhouse.client.ClickHouseProtocol;
 import com.clickhouse.client.ClickHouseServerForTest;
 import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.ClientConfigProperties;
 import com.clickhouse.client.api.ClientException;
 import com.clickhouse.client.api.DataTypeUtils;
 import com.clickhouse.client.api.command.CommandResponse;
@@ -269,6 +270,33 @@ public void insertRawData() throws Exception {
         assertEquals(records.size(), 1000);
     }
 
+    @Test(groups = { "integration" }, enabled = true)
+    public void insertRawDataAsync() throws Exception {
+        final String tableName = "raw_data_table_async";
+        final String createSQL = "CREATE TABLE " + tableName +
+                " (Id UInt32, event_ts Timestamp, name String, p1 Int64, p2 String) ENGINE = MergeTree() ORDER BY ()";
+
+        initTable(tableName, createSQL);
+
+        InsertSettings localSettings = new InsertSettings(settings.getAllSettings());
+        localSettings.setOption(ClientConfigProperties.ASYNC_OPERATIONS.getKey(), true);
+        ByteArrayOutputStream data = new ByteArrayOutputStream();
+        PrintWriter writer = new PrintWriter(data);
+        for (int i = 0; i < 1000; i++) {
+            writer.printf("%d\t%s\t%s\t%d\t%s\n", i, "2021-01-01 00:00:00", "name" + i, i, "p2");
+        }
+        writer.flush();
+        client.insert(tableName, new ByteArrayInputStream(data.toByteArray()),
+                ClickHouseFormat.TSV, localSettings).whenComplete((response, throwable) -> {
+                OperationMetrics metrics = response.getMetrics();
+                assertEquals((int)response.getWrittenRows(), 1000 );
+
+                List<GenericRecord> records = client.queryAll("SELECT * FROM " + tableName);
+                assertEquals(records.size(), 1000);
+        })
+                .join(); // wait operation complete. only for tests
+    }
+
     @Test(groups = { "integration" }, dataProvider = "insertRawDataSimpleDataProvider", dataProviderClass = InsertTests.class)
     public void insertRawDataSimple(String tableName) throws Exception {
 //        final String tableName = "raw_data_table";
@@ -639,10 +667,9 @@ public void testCollectionInsert() throws Exception {
         }
     }
 
-
-    static {
-        System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "DEBUG");
-    }
+//    static {
+//        System.setProperty("org.slf4j.simpleLogger.defaultLogLevel", "DEBUG");
+//    }
 
     @Test(groups = {"integration"}, dataProvider = "testAppCompressionDataProvider", dataProviderClass = InsertTests.class)
     public void testAppCompression(String algo) throws Exception {