Skip to content

Commit

Permalink
Multi-endpoint support (#956)
Browse files Browse the repository at this point in the history 
* Multi-endpoint support

* Skip one more test

* Add nullAsDefault JDBC option for handling null values

* Fix compile error on JDK 8

* More test for nullAsDefault option

* fix issues when fail over to different protocol or connecting to single-node

* Failover on UnknownHostException

* Correct exception handling

* Correct words

* support slash in connection parameter and case-insensitive enum value
  • Loading branch information
@zhicwu
zhicwu authored Jun 22, 2022
1 parent 425bb85 commit 75ddd37
Show file tree
Hide file tree
Showing 93 changed files with 5,316 additions and 1,274 deletions.
28 changes: 1 addition & 27 deletions clickhouse-benchmark/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
<scope>provided</scope>
</dependency>

<!-- JDBC drivers -->
<dependency>
<groupId>ru.yandex.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
Expand All @@ -49,33 +50,6 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
<artifactId>clickhouse-client</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>${project.parent.groupId}</groupId>
<artifactId>clickhouse-grpc-client</artifactId>
<version>${revision}</version>
<!-- exclusions>
<exclusion>
<groupId>${project.parent.groupId}</groupId>
<artifactId>io.grpc</artifactId>
</exclusion>
</exclusions -->
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
</dependency>
<!-- separate classifier did not work well with flatten plugin -->
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-okhttp</artifactId>
</dependency>

<!-- JDBC drivers -->
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
Expand Down
15 changes: 10 additions & 5 deletions clickhouse-cli-client/src/main/java/com/clickhouse/client/cli/ClickHouseCommandLine.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ public class ClickHouseCommandLine implements AutoCloseable {
private static final Logger log = LoggerFactory.getLogger(ClickHouseCommandLine.class);

public static final String DEFAULT_CLI_ARG_VERSION = "--version";
public static final String DEFAULT_CLICKHOUSE_CLI_PATH = "clickhouse-client";
public static final String DEFAULT_CLICKHOUSE_CLI_PATH = "clickhouse";
public static final String DEFAULT_CLIENT_OPTION = "client";
public static final String DEFAULT_DOCKER_CLI_PATH = "docker";
public static final String DEFAULT_DOCKER_IMAGE = "clickhouse/clickhouse-server";

Expand Down Expand Up @@ -103,9 +104,11 @@ static void dockerCommand(ClickHouseConfig config, String hostDir, String contai
}
String str = (String) config.getOption(ClickHouseCommandLineOption.CLI_CONTAINER_ID);
if (!ClickHouseChecker.isNullOrBlank(str)) {
if (!check(timeout, cli, "exec", str, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLI_ARG_VERSION)) {
if (!check(timeout, cli, "exec", str, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLIENT_OPTION,
DEFAULT_CLI_ARG_VERSION)) {
synchronized (ClickHouseCommandLine.class) {
if (!check(timeout, cli, "exec", str, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLI_ARG_VERSION)
if (!check(timeout, cli, "exec", str, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLIENT_OPTION,
DEFAULT_CLI_ARG_VERSION)
&& !check(timeout, cli, "run", "--rm", "--name", str, "-v", hostDir + ':' + containerDir,
"-d", img, "tail", "-f", "/dev/null")) {
throw new IllegalStateException("Failed to start new container: " + str);
Expand All @@ -117,7 +120,8 @@ static void dockerCommand(ClickHouseConfig config, String hostDir, String contai
commands.add("-i");
commands.add(str);
} else { // create new container for each query
if (!check(timeout, cli, "run", "--rm", img, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLI_ARG_VERSION)) {
if (!check(timeout, cli, "run", "--rm", img, DEFAULT_CLICKHOUSE_CLI_PATH, DEFAULT_CLIENT_OPTION,
DEFAULT_CLI_ARG_VERSION)) {
throw new IllegalStateException("Invalid ClickHouse docker image: " + img);
}
commands.add("run");
Expand Down Expand Up @@ -150,13 +154,14 @@ static Process startProcess(ClickHouseNode server, ClickHouseRequest<?> request)
if (ClickHouseChecker.isNullOrBlank(cli)) {
cli = DEFAULT_CLICKHOUSE_CLI_PATH;
}
if (!check(timeout, cli, DEFAULT_CLI_ARG_VERSION)) {
if (!check(timeout, cli, DEFAULT_CLIENT_OPTION, DEFAULT_CLI_ARG_VERSION)) {
// fallback to docker
dockerCommand(config, hostDir, containerDir, timeout, commands);
} else {
commands.add(cli);
containerDir = hostDir;
}
commands.add(DEFAULT_CLIENT_OPTION);

if (config.isSsl()) {
commands.add("--secure");
Expand Down
72 changes: 30 additions & 42 deletions ...house-cli-client/src/main/java/com/clickhouse/client/cli/ClickHouseCommandLineClient.java
View file
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
package com.clickhouse.client.cli;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;

import com.clickhouse.client.AbstractClient;
import com.clickhouse.client.ClickHouseChecker;
Expand All @@ -24,6 +25,19 @@
public class ClickHouseCommandLineClient extends AbstractClient<ClickHouseCommandLine> {
private static final Logger log = LoggerFactory.getLogger(ClickHouseCommandLineClient.class);

static final List<ClickHouseProtocol> SUPPORTED = Collections.singletonList(ClickHouseProtocol.TCP);

@Override
protected boolean checkHealth(ClickHouseNode server, int timeout) {
try (ClickHouseCommandLine cli = getConnection(connect(server).query("select 1"));
ClickHouseCommandLineResponse response = new ClickHouseCommandLineResponse(getConfig(), cli)) {
return response.firstRecord().getValue(0).asInteger() == 1;
} catch (Exception e) {
// ignore
}
return false;
}

@Override
protected ClickHouseCommandLine newConnection(ClickHouseCommandLine conn, ClickHouseNode server,
ClickHouseRequest<?> request) {
Expand All @@ -49,6 +63,16 @@ protected void closeConnection(ClickHouseCommandLine conn, boolean force) {
}
}

@Override
protected Collection<ClickHouseProtocol> getSupportedProtocols() {
return SUPPORTED;
}

@Override
protected ClickHouseResponse send(ClickHouseRequest<?> sealedRequest) throws ClickHouseException, IOException {
return new ClickHouseCommandLineResponse(sealedRequest.getConfig(), getConnection(sealedRequest));
}

@Override
public boolean accept(ClickHouseProtocol protocol) {
ClickHouseConfig config = getConfig();
Expand All @@ -65,50 +89,14 @@ public boolean accept(ClickHouseProtocol protocol) {
docker = ClickHouseCommandLine.DEFAULT_DOCKER_CLI_PATH;
}
return ClickHouseProtocol.TCP == protocol
&& (ClickHouseCommandLine.check(timeout, cli, ClickHouseCommandLine.DEFAULT_CLI_ARG_VERSION)
|| ClickHouseCommandLine.check(timeout, docker, ClickHouseCommandLine.DEFAULT_CLI_ARG_VERSION));
}

@Override
public CompletableFuture<ClickHouseResponse> execute(ClickHouseRequest<?> request) {
final ClickHouseRequest<?> sealedRequest = request.seal();
final ClickHouseConfig config = sealedRequest.getConfig();
final ClickHouseNode server = getServer();

if (config.isAsync()) {
return CompletableFuture
.supplyAsync(() -> {
try {
return new ClickHouseCommandLineResponse(config, getConnection(sealedRequest));
} catch (IOException e) {
throw new CompletionException(e);
}
});
} else {
try {
return CompletableFuture
.completedFuture(new ClickHouseCommandLineResponse(config, getConnection(sealedRequest)));
} catch (IOException e) {
throw new CompletionException(ClickHouseException.of(e, server));
}
}
&& (ClickHouseCommandLine.check(timeout, cli, ClickHouseCommandLine.DEFAULT_CLIENT_OPTION,
ClickHouseCommandLine.DEFAULT_CLI_ARG_VERSION)
|| ClickHouseCommandLine.check(timeout, docker, ClickHouseCommandLine.DEFAULT_CLIENT_OPTION,
ClickHouseCommandLine.DEFAULT_CLI_ARG_VERSION));
}

@Override
public final Class<? extends ClickHouseOption> getOptionClass() {
return ClickHouseCommandLineOption.class;
}

@Override
public boolean ping(ClickHouseNode server, int timeout) {
if (server != null) {
try (ClickHouseCommandLine cli = getConnection(connect(server).query("select 1"));
ClickHouseCommandLineResponse response = new ClickHouseCommandLineResponse(getConfig(), cli)) {
return response.firstRecord().getValue(0).asInteger() == 1;
} catch (Exception e) {
// ignore
}
}
return false;
}
}
26 changes: 19 additions & 7 deletions ...e-cli-client/src/test/java/com/clickhouse/client/cli/ClickHouseCommandLineClientTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,12 @@ protected ClickHouseNode getServer() {
return super.getServer();
}

@Test(groups = { "integration" })
@Override
public void testCustomLoad() throws Exception {
throw new SkipException("Skip due to time out error");
}

@Test(groups = { "integration" })
@Override
public void testLoadRawData() throws Exception {
Expand All @@ -56,17 +62,23 @@ public void testLoadRawData() throws Exception {
@Test(groups = { "integration" })
@Override
public void testMultipleQueries() throws Exception {
// FIXME not sure if the occasional "Stream closed" exception is related to zeroturnaround/zt-exec#30 or not
// FIXME not sure if the occasional "Stream closed" exception is related to
// zeroturnaround/zt-exec#30 or not
/*
Caused by: java.io.IOException: Stream closed
at java.io.BufferedInputStream.getBufIfOpen(BufferedInputStream.java:170)
at java.io.BufferedInputStream.read(BufferedInputStream.java:336)
at com.clickhouse.client.stream.WrappedInputStream.updateBuffer(WrappedInputStream.java:32)
at com.clickhouse.client.stream.AbstractByteArrayInputStream.available(AbstractByteArrayInputStream.java:56)
at com.clickhouse.client.ClickHouseDataProcessor.hasNext(ClickHouseDataProcessor.java:126)
* Caused by: java.io.IOException: Stream closed
* at java.io.BufferedInputStream.getBufIfOpen(BufferedInputStream.java:170)
* at java.io.BufferedInputStream.read(BufferedInputStream.java:336)
* at com.clickhouse.client.stream.WrappedInputStream.updateBuffer(
* WrappedInputStream.java:32)
* at com.clickhouse.client.stream.AbstractByteArrayInputStream.available(
* AbstractByteArrayInputStream.java:56)
* at
* com.clickhouse.client.ClickHouseDataProcessor.hasNext(ClickHouseDataProcessor
* .java:126)
*/
throw new SkipException("Skip due to unknown cause");
}

@Test(groups = { "integration" })
@Override
public void testReadWriteGeoTypes() {
Expand Down
Loading

0 comments on commit 75ddd37

Please sign in to comment.