Merge pull request #2880 from kishansinghifs1/feature/getNextAliveNode

Feat: getNextAliveNode() to failover to alternate endpoints on retryable failures.

Author: chernser

CHANGELOG.md

@@ -149,6 +149,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)

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

@@ -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;
@@ -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;
@@ -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,
@@ -187,9 +188,8 @@ private Client(Collection<Endpoint> endpoints, Map<String,String> configuration,
         }
 
         this.endpoints = tmpEndpoints.build();
+        this.nodeSelector = new ClientNodeSelector(this.endpoints);
 
-        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();
@@ -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()) {
@@ -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) {
@@ -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(),
@@ -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), "{}");
@@ -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);        };
 
@@ -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(),
@@ -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);
@@ -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);
         };
@@ -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
@@ -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);
             };
@@ -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 = ",";
 
@@ -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>

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

@@ -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

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -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) {
+                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);

client-v2/src/main/java/com/clickhouse/client/api/transport/ClientNodeSelector.java

@@ -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();
+    }
+}

client-v2/src/main/java/com/clickhouse/client/api/transport/EndpointState.java

@@ -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) {
+        if (quarantineMs <= 0) {
+            throw new IllegalArgumentException("Quarantine duration must be positive: " + quarantineMs);
+        }
+        this.failedUntil = System.currentTimeMillis() + quarantineMs;
+    }
+
+    boolean isAlive() {
+        return System.currentTimeMillis() >= failedUntil;
+    }
+}