Merge pull request #1674 from ClickHouse/adjusting-kafka-feedback

Adding proxy support, adding some debug methods, tweaking styles based on Kafka Connect rework

Author: Paultagoras

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

@@ -11,6 +11,8 @@
 import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
 import com.clickhouse.client.api.data_formats.RowBinaryWithNamesAndTypesFormatReader;
 import com.clickhouse.client.api.data_formats.internal.MapBackedRecord;
+import com.clickhouse.client.api.enums.Protocol;
+import com.clickhouse.client.api.enums.ProxyType;
 import com.clickhouse.client.api.insert.DataSerializationException;
 import com.clickhouse.client.api.insert.InsertResponse;
 import com.clickhouse.client.api.insert.InsertSettings;
@@ -46,6 +48,7 @@
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -187,12 +190,12 @@ public Builder addEndpoint(String endpoint) {
          * @param host - Endpoint host
          * @param port - Endpoint port
          */
-        public Builder addEndpoint(Protocol protocol, String host, int port) {
+        public Builder addEndpoint(Protocol protocol, String host, int port, boolean secure) {
             ValidationUtils.checkNonBlank(host, "host");
             ValidationUtils.checkNotNull(protocol, "protocol");
             ValidationUtils.checkRange(port, 1, ValidationUtils.TCP_PORT_NUMBER_MAX, "port");
 
-            String endpoint = String.format("%s://%s:%d", protocol.toString().toLowerCase(), host, port);
+            String endpoint = String.format("%s%s://%s:%d", protocol.toString().toLowerCase(), secure ? "s": "", host, port);
             this.addEndpoint(endpoint);
             return this;
         }
@@ -377,6 +380,17 @@ public Builder setDefaultDatabase(String database) {
             return this;
         }
 
+        public Builder addProxy(ProxyType type, String host, int port) {
+            ValidationUtils.checkNotNull(type, "type");
+            ValidationUtils.checkNonBlank(host, "host");
+            ValidationUtils.checkRange(port, 1, ValidationUtils.TCP_PORT_NUMBER_MAX, "port");
+
+            this.configuration.put(String.valueOf(ClickHouseClientOption.PROXY_TYPE), type.toString());
+            this.configuration.put(String.valueOf(ClickHouseClientOption.PROXY_HOST), host);
+            this.configuration.put(String.valueOf(ClickHouseClientOption.PROXY_PORT), String.valueOf(port));
+            return this;
+        }
+
         public Client build() {
             // check if endpoint are empty. so can not initiate client
             if (this.endpoints.isEmpty()) {
@@ -574,10 +588,7 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
      * @param format - format of the data in the stream
      * @return {@code CompletableFuture<InsertResponse>} - a promise to insert response
      */
-    public CompletableFuture<InsertResponse> insert(String tableName,
-                                         InputStream data,
-                                         ClickHouseFormat format)
-    {
+    public CompletableFuture<InsertResponse> insert(String tableName, InputStream data, ClickHouseFormat format) {
         return insert(tableName, data, format, new InsertSettings());
     }
 
@@ -867,4 +878,18 @@ private String startOperation() {
         globalClientStats.put(operationId, new ClientStatisticsHolder());
         return operationId;
     }
+
+    public String toString() {
+        return "Client{" +
+                "endpoints=" + endpoints +
+                '}';
+    }
+
+    public Map<String, String> getConfiguration() {
+        return Collections.unmodifiableMap(configuration);
+    }
+
+    public Set<String> getEndpoints() {
+        return Collections.unmodifiableSet(endpoints);
+    }
 }

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

@@ -0,0 +1,7 @@
+package com.clickhouse.client.api.enums;
+
+public enum Protocol {
+    HTTP,
+    //HTTPS,
+    //TCP,
+}

client-v2/src/main/java/com/clickhouse/client/api/enums/Protocol.java

@@ -0,0 +1,7 @@
+package com.clickhouse.client.api.enums;
+
+public enum ProxyType {
+    //DIRECT,
+    HTTP,
+    SOCKS;
+}

client-v2/src/main/java/com/clickhouse/client/api/enums/ProxyType.java

@@ -8,7 +8,7 @@
 import com.clickhouse.client.ClickHouseNodeSelector;
 import com.clickhouse.client.ClickHouseProtocol;
 import com.clickhouse.client.api.Client;
-import com.clickhouse.client.api.Protocol;
+import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.insert.InsertResponse;
 import com.clickhouse.client.api.insert.InsertSettings;
 import com.clickhouse.client.api.metrics.ClientMetrics;
@@ -34,7 +34,7 @@ public class InsertTests extends BaseIntegrationTest {
     public void setUp() {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
         client = new Client.Builder()
-                .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort())
+                .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), false)
                 .setUsername("default")
                 .setPassword("")
                 .build();

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

@@ -12,7 +12,7 @@
 import com.clickhouse.client.api.Client;
 import com.clickhouse.client.api.ClientException;
 import com.clickhouse.client.api.DataTypeUtils;
-import com.clickhouse.client.api.Protocol;
+import com.clickhouse.client.api.enums.Protocol;
 import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader;
 import com.clickhouse.client.api.data_formats.NativeFormatReader;
 import com.clickhouse.client.api.data_formats.RowBinaryFormatReader;
@@ -74,7 +74,7 @@ public class QueryTests extends BaseIntegrationTest {
     public void setUp() {
         ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
         client = new Client.Builder()
-                .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort())
+                .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), false)
                 .setUsername("default")
                 .setPassword("")
                 .build();