Skip to content
Open
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,8 @@ like `ch_db_01`. This is mostly used in k8s environment. (https://github.com/Cli

- **[repo]** Added a contribution guide. Please review and send us your feedback. (https://github.com/ClickHouse/clickhouse-java/pull/2859)

- **[client-v2]** Added endpoint failover support: when multiple endpoints are configured and a request fails with a retryable error (connect timeout, connection refused, HTTP 503, etc.), the client now automatically retries against the next available endpoint instead of always targeting the first one. Failed endpoints are quarantined for 30 seconds before being retried. (https://github.com/ClickHouse/clickhouse-java/issues/2855)

### Bug Fixes

- **[jdbc-v2, client-v2]** Fixed error handling for responses that not a ClickHouse error, like `404` response. (https://github.com/ClickHouse/clickhouse-java/issues/2803)
Expand Down
95 changes: 43 additions & 52 deletions client-v2/src/main/java/com/clickhouse/client/api/Client.java
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import com.clickhouse.client.api.serde.POJOFieldDeserializer;
import com.clickhouse.client.api.serde.POJOFieldSerializer;
import com.clickhouse.client.api.serde.POJOSerDe;
import com.clickhouse.client.api.transport.ClientNodeSelector;
import com.clickhouse.client.api.transport.Endpoint;
import com.clickhouse.client.api.transport.HttpEndpoint;
import com.clickhouse.client.config.ClickHouseClientOption;
Expand All @@ -62,8 +63,8 @@
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -140,9 +141,9 @@ public class Client implements AutoCloseable {
private String dbUser;
private String serverVersion;
private final Object metricsRegistry;
private final int retries;
private LZ4Factory lz4Factory = null;
private final Supplier<String> queryIdGenerator;
private final ClientNodeSelector nodeSelector;
private final CredentialsManager credentialsManager;

private Client(Collection<Endpoint> endpoints, Map<String,String> configuration,
Expand Down Expand Up @@ -187,9 +188,8 @@ private Client(Collection<Endpoint> endpoints, Map<String,String> configuration,
}

this.endpoints = tmpEndpoints.build();
this.nodeSelector = new ClientNodeSelector(this.endpoints);
Comment thread
cursor[bot] marked this conversation as resolved.

String retry = configuration.get(ClientConfigProperties.RETRY_ON_FAILURE.getKey());
this.retries = retry == null ? 0 : Integer.parseInt(retry);
boolean useNativeCompression = !MapUtils.getFlag(configuration, ClientConfigProperties.DISABLE_NATIVE_COMPRESSION.getKey(), false);
if (useNativeCompression) {
this.lz4Factory = LZ4Factory.fastestInstance();
Expand Down Expand Up @@ -272,7 +272,7 @@ public static class Builder {
private Supplier<String> queryIdGenerator;

public Builder() {
this.endpoints = new HashSet<>();
this.endpoints = new LinkedHashSet<>();
this.configuration = new HashMap<>();

for (ClientConfigProperties p : ClientConfigProperties.values()) {
Expand Down Expand Up @@ -1364,8 +1364,8 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
}


Integer retry = (Integer) configuration.get(ClientConfigProperties.RETRY_ON_FAILURE.getKey());
final int maxRetries = retry == null ? 0 : retry;
final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings());
final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1);

requestSettings.setOption(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey(), format);
if (requestSettings.getQueryId() == null && queryIdGenerator != null) {
Expand All @@ -1374,10 +1374,10 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
Supplier<InsertResponse> supplier = () -> {
long startTime = System.nanoTime();
// Selecting some node
Endpoint selectedEndpoint = getNextAliveNode();
Endpoint selectedEndpoint = nodeSelector.getEndpoint();

RuntimeException lastException = null;
for (int i = 0; i <= maxRetries; i++) {
for (int i = 0; i <= maxAttempts; i++) {
// Execute request
try (ClassicHttpResponse httpResponse =
httpClientHelper.executeRequest(selectedEndpoint, requestSettings.getAllSettings(),
Expand All @@ -1400,14 +1400,6 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
out.close();
})) {


// Check response
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
selectedEndpoint = getNextAliveNode();
continue;
}

ClientStatisticsHolder clientStats = globalClientStats.remove(operationId);
OperationMetrics metrics = new OperationMetrics(clientStats);
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}");
Expand All @@ -1420,15 +1412,19 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
LOG.warn("Retrying.", e);
selectedEndpoint = getNextAliveNode();
if (i < maxAttempts) {
LOG.warn("Retrying.", e);
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
} else {
nodeSelector.getNextAliveNode(selectedEndpoint);
}
} else {
throw lastException;
}
}
}

String errMsg = requestExMsg("Insert", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId());
String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
LOG.warn(errMsg);
throw (lastException == null ? new ClientException(errMsg) : lastException); };

Expand Down Expand Up @@ -1593,13 +1589,15 @@ public CompletableFuture<InsertResponse> insert(String tableName,
if (requestSettings.getQueryId() == null && queryIdGenerator != null) {
requestSettings.setQueryId(queryIdGenerator.get());
}
final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings());
final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1);
responseSupplier = () -> {
long startTime = System.nanoTime();
// Selecting some node
Endpoint selectedEndpoint = getNextAliveNode();
Endpoint selectedEndpoint = nodeSelector.getEndpoint();

RuntimeException lastException = null;
for (int i = 0; i <= retries; i++) {
for (int i = 0; i <= maxAttempts; i++) {
// Execute request
try (ClassicHttpResponse httpResponse =
httpClientHelper.executeRequest(selectedEndpoint, requestSettings.getAllSettings(),
Expand All @@ -1608,14 +1606,6 @@ public CompletableFuture<InsertResponse> insert(String tableName,
out.close();
})) {


// Check response
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
selectedEndpoint = getNextAliveNode();
continue;
}

OperationMetrics metrics = new OperationMetrics(finalClientStats);
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}");
ProcessParser.parseSummary(summary, metrics);
Expand All @@ -1627,22 +1617,26 @@ public CompletableFuture<InsertResponse> insert(String tableName,
String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
LOG.warn("Retrying.", e);
selectedEndpoint = getNextAliveNode();
if (i < maxAttempts) {
LOG.warn("Retrying.", e);
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
} else {
nodeSelector.getNextAliveNode(selectedEndpoint);
}
} else {
throw lastException;
}
}

if (i < retries) {
if (i < maxAttempts) {
try {
writer.onRetry();
} catch (IOException ioe) {
throw new ClientException("Failed to reset stream before next attempt", ioe);
}
}
}
String errMsg = requestExMsg("Insert", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId());
String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
LOG.warn(errMsg);
throw (lastException == null ? new ClientException(errMsg) : lastException);
};
Expand Down Expand Up @@ -1731,24 +1725,19 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
requestSettings.setQueryId(queryIdGenerator.get());
}

final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings());
final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1);
Supplier<QueryResponse> responseSupplier = () -> {
long startTime = System.nanoTime();
// Selecting some node
Endpoint selectedEndpoint = getNextAliveNode();
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
RuntimeException lastException = null;
for (int i = 0; i <= retries; i++) {
for (int i = 0; i <= maxAttempts; i++) {
ClassicHttpResponse httpResponse = null;
try {
httpResponse = httpClientHelper.executeRequest(selectedEndpoint,
requestSettings.getAllSettings(),
sqlQuery);
// Check response
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
selectedEndpoint = getNextAliveNode();
HttpAPIClientHelper.closeQuietly(httpResponse);
continue;
}
requestSettings.getAllSettings(),
sqlQuery);

OperationMetrics metrics = new OperationMetrics(clientStats);
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse
Expand All @@ -1772,14 +1761,18 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
String msg = requestExMsg("Query", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
LOG.warn("Retrying.", e);
selectedEndpoint = getNextAliveNode();
if (i < maxAttempts) {
LOG.warn("Retrying.", e);
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
} else {
nodeSelector.getNextAliveNode(selectedEndpoint);
}
} else {
throw lastException;
}
}
}
String errMsg = requestExMsg("Query", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId());
String errMsg = requestExMsg("Query", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
LOG.warn(errMsg);
throw (lastException == null ? new ClientException(errMsg) : lastException);
};
Expand Down Expand Up @@ -2285,9 +2278,7 @@ public void updateAccessToken(String accessToken) {
this.credentialsManager.setAccessToken(accessToken);
}

private Endpoint getNextAliveNode() {
return endpoints.get(0);
}


public static final String VALUES_LIST_DELIMITER = ",";

Expand All @@ -2313,7 +2304,7 @@ private Map<String, Object> buildRequestSettings(Map<String, Object> opSettings)
* <p>For {@link ClickHouseFormat#JSONEachRow}, callers may opt in to plain JSON numbers by setting
* {@link ClientConfigProperties#JSON_DISABLE_NUMBER_QUOTING}. Explicit server settings are otherwise
* left untouched.</p>
* <ul>
* <ul>
* <li>{@code output_format_json_quote_64bit_integers}</li>
* <li>{@code output_format_json_quote_64bit_floats}</li>
* <li>{@code output_format_json_quote_decimals}</li>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,8 @@ public String getQueryId() {
}

private boolean discoverIsRetryable(int code, String message, int transportProtocolCode) {
//Let's check if we have a ServerException to reference the error code
//https://github.com/ClickHouse/ClickHouse/blob/master/src/Common/ErrorCodes.cpp
// Let's check if we have a ServerException to reference the error code
// https://github.com/ClickHouse/ClickHouse/blob/master/src/Common/ErrorCodes.cpp
switch (code) { // UNEXPECTED_END_OF_FILE
case 3: // UNEXPECTED_END_OF_FILE
case 107: // FILE_DOESNT_EXIST
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -594,7 +594,10 @@ private ClassicHttpResponse doPostRequest(Map<String, Object> requestConfig, Htt
throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings.");
} else if (httpResponse.getCode() == HttpStatus.SC_BAD_GATEWAY) {
httpResponse.close();
throw new ClientException("Server returned '502 Bad gateway'. Check network and proxy settings.");
throw new ConnectException("Server returned '502 Bad gateway'. Check network and proxy settings.");
} else if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
Comment thread
chernser marked this conversation as resolved.
httpResponse.close();
throw new ConnectException("Server returned '503 Service unavailable'.");
} else if (httpResponse.getCode() >= HttpStatus.SC_BAD_REQUEST || httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) {
try {
throw readError(req, httpResponse);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
package com.clickhouse.client.api.transport;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
* So we will look up for the first-alive node and then assign that node to
* client to talk.
*
* <p>Endpoints are tried in the order they were registered (index 0 is the
* "primary"). When a request fails, the failed endpoint is quarantined for
* {@link #DEFAULT_QUARANTINE_MS} milliseconds and the next alive endpoint
* is returned. Once the quarantine expires the node automatically becomes
* eligible again, so traffic returns to the primary without any explicit
* reset.</p>
*
* <p>If all endpoints are quarantined, the primary (index 0) is returned
* as a fallback to avoid a complete lockout.</p>
*
* <p>This class is thread-safe: concurrent callers may invoke
* {@link #getEndpoint()} and {@link #getNextAliveNode(Endpoint)}
* from different threads.</p>
*/
public class ClientNodeSelector {

private static final Logger LOG = LoggerFactory.getLogger(ClientNodeSelector.class);

static final long DEFAULT_QUARANTINE_MS = 30_000;

private final List<EndpointState> endpointStates;

public ClientNodeSelector(List<Endpoint> endpoints) {
List<EndpointState> states = new ArrayList<>(endpoints.size());
for (Endpoint ep : endpoints) {
states.add(new EndpointState(ep));
}
this.endpointStates = Collections.unmodifiableList(states);
}

public Endpoint getEndpoint() {
for (EndpointState state : endpointStates) {
if (state.isAlive()) {
return state.getEndpoint();
}
}
LOG.warn("All endpoints are non-responsive, falling back to primary endpoint");
return endpointStates.get(0).getEndpoint();
}

public Endpoint getNextAliveNode(Endpoint failedEndpoint) {
for (EndpointState state : endpointStates) {
if (state.getEndpoint().equals(failedEndpoint)) {
state.markFailed(DEFAULT_QUARANTINE_MS);
LOG.warn("Endpoint {} quarantined for {} ms", failedEndpoint.getHost(), DEFAULT_QUARANTINE_MS);
break;
}
}
return getEndpoint();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
package com.clickhouse.client.api.transport;

/**
*{@link Endpoint} is wrapped to track for failover.
* When a node fails, it can be quarantined for a fixed duration and after
* it is expired, it will be considered as alive again.
*/
class EndpointState {

private final Endpoint endpoint;

private volatile long failedUntil;

EndpointState(Endpoint endpoint) {
this.endpoint = endpoint;
this.failedUntil = 0;
}

Endpoint getEndpoint() {
return endpoint;
}

void markFailed(long quarantineMs) {
Comment thread
chernser marked this conversation as resolved.
if (quarantineMs <= 0) {
throw new IllegalArgumentException("Quarantine duration must be positive: " + quarantineMs);
}
this.failedUntil = System.currentTimeMillis() + quarantineMs;
}

boolean isAlive() {
return System.currentTimeMillis() >= failedUntil;
}
}
Loading
Loading