diff --git a/CHANGELOG.md b/CHANGELOG.md index ab70ea343..2e680df8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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) diff --git a/client-v2/src/main/java/com/clickhouse/client/api/Client.java b/client-v2/src/main/java/com/clickhouse/client/api/Client.java index 3bf94c529..2066cff14 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/Client.java +++ b/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 queryIdGenerator; + private final ClientNodeSelector nodeSelector; private final CredentialsManager credentialsManager; private Client(Collection endpoints, Map configuration, @@ -187,9 +188,8 @@ private Client(Collection endpoints, Map 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 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 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 insert(String tableName, List data, Supplier 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 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 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 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 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,14 +1617,18 @@ public CompletableFuture 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) { @@ -1642,7 +1636,7 @@ public CompletableFuture insert(String tableName, } } } - 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 query(String sqlQuery, Map 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 query(String sqlQuery, Map buildRequestSettings(Map opSettings) *

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.

- *
    + *
      *
    • {@code output_format_json_quote_64bit_integers}
      • *
    • {@code output_format_json_quote_64bit_floats}
      • *
    • {@code output_format_json_quote_decimals}
      • diff --git a/client-v2/src/main/java/com/clickhouse/client/api/ServerException.java b/client-v2/src/main/java/com/clickhouse/client/api/ServerException.java index bd2361bfa..c324e30ea 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/ServerException.java +++ b/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 diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index ab4b0153c..ebefee9f3 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java @@ -594,7 +594,10 @@ private ClassicHttpResponse doPostRequest(Map 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); diff --git a/client-v2/src/main/java/com/clickhouse/client/api/transport/ClientNodeSelector.java b/client-v2/src/main/java/com/clickhouse/client/api/transport/ClientNodeSelector.java new file mode 100644 index 000000000..1fefff8f2 --- /dev/null +++ b/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. + * + *

      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.

      + * + *

      If all endpoints are quarantined, the primary (index 0) is returned + * as a fallback to avoid a complete lockout.

      + * + *

      This class is thread-safe: concurrent callers may invoke + * {@link #getEndpoint()} and {@link #getNextAliveNode(Endpoint)} + * from different threads.

      + */ +public class ClientNodeSelector { + + private static final Logger LOG = LoggerFactory.getLogger(ClientNodeSelector.class); + + static final long DEFAULT_QUARANTINE_MS = 30_000; + + private final List endpointStates; + + public ClientNodeSelector(List endpoints) { + List 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(); + } +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/transport/EndpointState.java b/client-v2/src/main/java/com/clickhouse/client/api/transport/EndpointState.java new file mode 100644 index 000000000..4671400be --- /dev/null +++ b/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; + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/ClientFailoverTest.java b/client-v2/src/test/java/com/clickhouse/client/ClientFailoverTest.java new file mode 100644 index 000000000..6a724151d --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/ClientFailoverTest.java @@ -0,0 +1,253 @@ +package com.clickhouse.client; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.ClientException; +import com.clickhouse.client.api.ClientFaultCause; +import com.clickhouse.client.api.enums.Protocol; +import com.clickhouse.client.api.query.GenericRecord; +import com.clickhouse.data.ClickHouseFormat; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * Integration tests for endpoint failover behavior in client-v2. + * + *

      Verifies that when multiple endpoints are configured and the first + * endpoint is unreachable, the client automatically fails over to the + * next available endpoint.

      + */ +public class ClientFailoverTest extends BaseIntegrationTest { + private static final Logger LOGGER = LoggerFactory.getLogger(ClientFailoverTest.class); + + /** + * Configures a dead endpoint (port 1, nothing listens) as the primary + * and the actual test server as the backup. Verifies that a query + * succeeds by failing over to the backup node. + */ + @Test(groups = {"integration"}) + public void testQueryFailoverToBackupNode() { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .retryOnFailures(ClientFaultCause.ConnectTimeout, ClientFaultCause.NoHttpResponse) + .setMaxRetries(3) + .build()) { + + List result = client.queryAll("SELECT 1 AS val"); + Assert.assertFalse(result.isEmpty(), "Expected at least one record"); + Assert.assertEquals(result.get(0).getInteger("val"), Integer.valueOf(1), "Query should succeed via failover to the backup node"); + } + } + + /** + * Verifies that when all endpoints are healthy, the primary (first) + * endpoint is consistently used. This tests the "affinity" behavior. + */ + @Test(groups = {"integration"}) + public void testPrimaryEndpointAffinityWhenHealthy() { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build()) { + + for (int i = 0; i < 5; i++) { + List result = client.queryAll("SELECT " + i + " AS val"); + Assert.assertFalse(result.isEmpty()); + Assert.assertEquals(result.get(0).getInteger("val"), Integer.valueOf(i)); + } + } + } + + /** + * Verifies that insert operations also failover when the primary + * endpoint is down. + */ + @Test(groups = {"integration"}) + public void testInsertFailoverToBackupNode() throws Exception { + if (isCloud()) { + return; + } + + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + + try (Client adminClient = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build()) { + adminClient.execute("DROP TABLE IF EXISTS failover_insert_test").get(10, TimeUnit.SECONDS).close(); + adminClient.execute("CREATE TABLE failover_insert_test (val UInt32) ENGINE MergeTree ORDER BY ()").get(10, TimeUnit.SECONDS).close(); + } + + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .retryOnFailures(ClientFaultCause.ConnectTimeout, ClientFaultCause.NoHttpResponse) + .setMaxRetries(3) + .build()) { + + String csvData = "42\n"; + client.insert("failover_insert_test", + new ByteArrayInputStream(csvData.getBytes()), + ClickHouseFormat.CSV).get(30, TimeUnit.SECONDS).close(); + } + + try (Client verifyClient = new Client.Builder() + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build()) { + List result = verifyClient.queryAll("SELECT val FROM failover_insert_test"); + Assert.assertFalse(result.isEmpty(), "Expected at least one row after failover insert"); + Assert.assertEquals(result.get(0).getInteger("val"), Integer.valueOf(42)); + + verifyClient.execute("DROP TABLE IF EXISTS failover_insert_test").get(10, TimeUnit.SECONDS).close(); + } + } + + @Test(groups = {"integration"}) + public void testMultipleBackups() { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") + .addEndpoint("http://127.0.0.1:2") + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) // working + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .retryOnFailures(ClientFaultCause.ConnectTimeout, ClientFaultCause.NoHttpResponse) + .setMaxRetries(3) + .build()) { + + List result = client.queryAll("SELECT 1 AS val"); + Assert.assertFalse(result.isEmpty()); + Assert.assertEquals(result.get(0).getInteger("val"), Integer.valueOf(1)); + } + } + + @Test(groups = {"integration"}, expectedExceptions = ClientException.class) + public void testAllEndpointsDead() { + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") + .addEndpoint("http://127.0.0.1:2") + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .retryOnFailures(ClientFaultCause.ConnectTimeout, ClientFaultCause.NoHttpResponse) + .setMaxRetries(3) + .build()) { + + client.queryAll("SELECT 1 AS val"); + } + } + + @Test(groups = {"integration"}, expectedExceptions = ClientException.class) + public void testRetryLimitReached() { + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") + .addEndpoint("http://127.0.0.1:2") + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .retryOnFailures(ClientFaultCause.ConnectTimeout, ClientFaultCause.NoHttpResponse) + .setMaxRetries(1) + .build()) { + + client.queryAll("SELECT 1 AS val"); + } + } + + @Test(groups = {"integration"}) + public void testDuplicateEndpointRegistrationAndOrderPreservation() throws Exception { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + + String uriA = "http://127.0.0.1:1"; + String uriB = "http://127.0.0.1:2"; + + try (Client client = new Client.Builder() + .addEndpoint(uriA) + .addEndpoint(uriA) + .addEndpoint(uriB) + .addEndpoint(uriB) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build()) { + + java.lang.reflect.Field field = Client.class.getDeclaredField("endpoints"); + field.setAccessible(true); + @SuppressWarnings("unchecked") + List endpoints = + (List) field.get(client); + + Assert.assertEquals(endpoints.size(), 2, "Should only have unique endpoints"); + Assert.assertEquals(endpoints.get(0).getURI().toString(), uriA + "/", "First endpoint should be " + uriA + "/"); + Assert.assertEquals(endpoints.get(1).getURI().toString(), uriB + "/", "Second endpoint should be " + uriB + "/"); + } + } + + @Test(groups = {"integration"}, expectedExceptions = IllegalArgumentException.class) + public void testNoEndpointsConfigured() { + new Client.Builder() + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .build(); + } + + @Test(groups = {"integration"}) + public void testHTTP503Failover() throws Exception { + ClickHouseNode node = getServer(ClickHouseProtocol.HTTP); + boolean isSecure = isCloud(); + + WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + mockServer.start(); + try { + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse().withStatus(503).withBody("Service Unavailable"))); + + try (Client client = new Client.Builder() + .addEndpoint("http://localhost:" + mockServer.port()) + .addEndpoint(Protocol.HTTP, node.getHost(), node.getPort(), isSecure) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .setMaxRetries(3) + .build()) { + + List result = client.queryAll("SELECT 1 AS val"); + Assert.assertFalse(result.isEmpty(), "Expected at least one record"); + Assert.assertEquals(result.get(0).getInteger("val"), Integer.valueOf(1), + "Query should succeed via failover after 503"); + } + } finally { + mockServer.stop(); + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/ClientFailoverUnitTest.java b/client-v2/src/test/java/com/clickhouse/client/api/ClientFailoverUnitTest.java new file mode 100644 index 000000000..fdffc68b9 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/ClientFailoverUnitTest.java @@ -0,0 +1,42 @@ +package com.clickhouse.client.api; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.query.QueryResponse; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.util.concurrent.TimeUnit; + +public class ClientFailoverUnitTest { + + @Test + public void testWireMockFailoverOnly() throws Exception { + WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort()); + mockServer.start(); + try { + mockServer.stubFor(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse().withStatus(200) + .withHeader("Content-Type", "text/plain") + .withBody(""))); + + try (Client client = new Client.Builder() + .addEndpoint("http://127.0.0.1:1") // dead endpoint + .addEndpoint("http://localhost:" + mockServer.port()) // healthy mock endpoint + .setUsername("default") + .setPassword("password") + .setDefaultDatabase("default") + .setMaxRetries(3) + .build()) { + + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + } finally { + mockServer.stop(); + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java index f03e84af6..90f3be551 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/HttpAPIClientHelperTest.java @@ -1,5 +1,78 @@ -package com.clickhouse.client.api.internal; - -public class HttpAPIClientHelperTest { - +package com.clickhouse.client.api.internal; + +import com.clickhouse.client.api.transport.Endpoint; +import com.clickhouse.client.api.transport.HttpEndpoint; +import net.jpountz.lz4.LZ4Factory; +import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; +import org.apache.hc.core5.http.ClassicHttpResponse; +import org.apache.hc.core5.http.HttpEntity; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.lang.reflect.Field; +import java.net.ConnectException; +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class HttpAPIClientHelperTest { + + @Test + public void testExecuteRequestThrowsConnectExceptionOn502() throws Exception { + Map configuration = new HashMap<>(); + HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance()); + + CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class); + Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient"); + httpClientField.setAccessible(true); + httpClientField.set(helper, mockHttpClient); + + ClassicHttpResponse mockResponse = mock(ClassicHttpResponse.class); + when(mockResponse.getCode()).thenReturn(502); + HttpEntity mockEntity = mock(HttpEntity.class); + when(mockResponse.getEntity()).thenReturn(mockEntity); + + when(mockHttpClient.executeOpen(any(), any(), any())).thenReturn(mockResponse); + + Endpoint endpoint = new HttpEndpoint("localhost", 8123, false, "/"); + try { + helper.executeRequest(endpoint, new HashMap<>(), "SELECT 1"); + Assert.fail("Expected ConnectException to be thrown"); + } catch (ConnectException e) { + // expected + } catch (Exception e) { + Assert.fail("Expected ConnectException to be thrown, but got: " + e.getClass().getName(), e); + } + } + + @Test + public void testExecuteRequestThrowsConnectExceptionOn503() throws Exception { + Map configuration = new HashMap<>(); + HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance()); + + CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class); + Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient"); + httpClientField.setAccessible(true); + httpClientField.set(helper, mockHttpClient); + + ClassicHttpResponse mockResponse = mock(ClassicHttpResponse.class); + when(mockResponse.getCode()).thenReturn(503); + HttpEntity mockEntity = mock(HttpEntity.class); + when(mockResponse.getEntity()).thenReturn(mockEntity); + + when(mockHttpClient.executeOpen(any(), any(), any())).thenReturn(mockResponse); + + Endpoint endpoint = new HttpEndpoint("localhost", 8123, false, "/"); + try { + helper.executeRequest(endpoint, new HashMap<>(), "SELECT 1"); + Assert.fail("Expected ConnectException to be thrown"); + } catch (ConnectException e) { + // expected + } catch (Exception e) { + Assert.fail("Expected ConnectException to be thrown, but got: " + e.getClass().getName(), e); + } + } } \ No newline at end of file diff --git a/client-v2/src/test/java/com/clickhouse/client/api/transport/ClientNodeSelectorTest.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/ClientNodeSelectorTest.java new file mode 100644 index 000000000..b973c48db --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/ClientNodeSelectorTest.java @@ -0,0 +1,133 @@ +package com.clickhouse.client.api.transport; + +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class ClientNodeSelectorTest { + + @Test + public void testEndpointStateQuarantineAndExpiry() throws InterruptedException { + Endpoint ep = new HttpEndpoint("localhost", 8123, false, "/"); + EndpointState state = new EndpointState(ep); + + Assert.assertTrue(state.isAlive(), "New endpoint state should be alive"); + Assert.assertSame(state.getEndpoint(), ep); + + state.markFailed(20); + Assert.assertFalse(state.isAlive(), "Endpoint state should not be alive immediately after failure"); + + Thread.sleep(40); + Assert.assertTrue(state.isAlive(), "Endpoint state should be alive again after quarantine expires"); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testEndpointStateQuarantineValidation() { + Endpoint ep = new HttpEndpoint("localhost", 8123, false, "/"); + EndpointState state = new EndpointState(ep); + state.markFailed(0); + } + + @Test(expectedExceptions = IllegalArgumentException.class) + public void testEndpointStateQuarantineValidationNegative() { + Endpoint ep = new HttpEndpoint("localhost", 8123, false, "/"); + EndpointState state = new EndpointState(ep); + state.markFailed(-10); + } + + @Test + public void testClientNodeSelectorAffinityAndQuarantine() { + Endpoint epA = new HttpEndpoint("localhost", 8123, false, "/"); + Endpoint epB = new HttpEndpoint("localhost", 8124, false, "/"); + Endpoint epC = new HttpEndpoint("localhost", 8125, false, "/"); + + ClientNodeSelector selector = new ClientNodeSelector(Arrays.asList(epA, epB, epC)); + + Assert.assertEquals(selector.getEndpoint(), epA); + + Endpoint next = selector.getNextAliveNode(epA); + Assert.assertEquals(next, epB); + Assert.assertEquals(selector.getEndpoint(), epB); + + next = selector.getNextAliveNode(epB); + Assert.assertEquals(next, epC); + Assert.assertEquals(selector.getEndpoint(), epC); + + next = selector.getNextAliveNode(epC); + Assert.assertEquals(next, epA); + Assert.assertEquals(selector.getEndpoint(), epA); + } + + @Test + public void testClientNodeSelectorFallbackWhenAllDead() { + Endpoint epA = new HttpEndpoint("localhost", 8123, false, "/"); + Endpoint epB = new HttpEndpoint("localhost", 8124, false, "/"); + + ClientNodeSelector selector = new ClientNodeSelector(Arrays.asList(epA, epB)); + + selector.getNextAliveNode(epA); + selector.getNextAliveNode(epB); + + Assert.assertEquals(selector.getEndpoint(), epA); + } + + @Test + public void testClientNodeSelectorSequentialFailures() { + Endpoint epA = new HttpEndpoint("localhost", 8123, false, "/"); + Endpoint epB = new HttpEndpoint("localhost", 8124, false, "/"); + Endpoint epC = new HttpEndpoint("localhost", 8125, false, "/"); + + ClientNodeSelector selector = new ClientNodeSelector(Arrays.asList(epA, epB, epC)); + + Assert.assertEquals(selector.getEndpoint(), epA); + + Assert.assertEquals(selector.getNextAliveNode(epA), epB); + Assert.assertEquals(selector.getNextAliveNode(epB), epC); + Assert.assertEquals(selector.getNextAliveNode(epC), epA); + } + + @Test + public void testClientNodeSelectorConcurrency() throws Exception { + Endpoint epA = new HttpEndpoint("localhost", 8123, false, "/"); + Endpoint epB = new HttpEndpoint("localhost", 8124, false, "/"); + Endpoint epC = new HttpEndpoint("localhost", 8125, false, "/"); + + ClientNodeSelector selector = new ClientNodeSelector(Arrays.asList(epA, epB, epC)); + + int threadCount = 10; + int loopCount = 1000; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + List> futures = new ArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + futures.add(executor.submit(new Callable() { + @Override + public Void call() throws Exception { + for (int j = 0; j < loopCount; j++) { + Endpoint ep = selector.getEndpoint(); + Assert.assertNotNull(ep); + if (j % 10 == 0) { + selector.getNextAliveNode(ep); + } + } + return null; + } + })); + } + + executor.shutdown(); + Assert.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + + for (Future future : futures) { + future.get(); + } + } +}