Skip to content

Commit 512c0e0

Browse files
kishansinghifs1kishansinghifs1
committed
fix: calculate maximum retry attempts based on endpoint count and add failover unit test
1 parent 21f71f5 commit 512c0e0

2 files changed

Lines changed: 76 additions & 24 deletions

File tree

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

Lines changed: 34 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1367,8 +1367,8 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13671367
}
13681368

13691369

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

13731373
requestSettings.setOption(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey(), format);
13741374
if (requestSettings.getQueryId() == null && queryIdGenerator != null) {
@@ -1380,7 +1380,7 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13801380
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
13811381

13821382
RuntimeException lastException = null;
1383-
for (int i = 0; i <= maxRetries; i++) {
1383+
for (int i = 0; i <= maxAttempts; i++) {
13841384
// Execute request
13851385
try (ClassicHttpResponse httpResponse =
13861386
httpClientHelper.executeRequest(selectedEndpoint, requestSettings.getAllSettings(),
@@ -1415,15 +1415,19 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
14151415
String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
14161416
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
14171417
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
1418-
LOG.warn("Retrying.", e);
1419-
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1418+
if (i < maxAttempts) {
1419+
LOG.warn("Retrying.", e);
1420+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1421+
} else {
1422+
nodeSelector.getNextAliveNode(selectedEndpoint);
1423+
}
14201424
} else {
14211425
throw lastException;
14221426
}
14231427
}
14241428
}
14251429

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

@@ -1588,13 +1592,15 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15881592
if (requestSettings.getQueryId() == null && queryIdGenerator != null) {
15891593
requestSettings.setQueryId(queryIdGenerator.get());
15901594
}
1595+
final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings());
1596+
final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1);
15911597
responseSupplier = () -> {
15921598
long startTime = System.nanoTime();
15931599
// Selecting some node
15941600
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
15951601

15961602
RuntimeException lastException = null;
1597-
for (int i = 0; i <= retries; i++) {
1603+
for (int i = 0; i <= maxAttempts; i++) {
15981604
// Execute request
15991605
try (ClassicHttpResponse httpResponse =
16001606
httpClientHelper.executeRequest(selectedEndpoint, requestSettings.getAllSettings(),
@@ -1614,22 +1620,26 @@ public CompletableFuture<InsertResponse> insert(String tableName,
16141620
String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
16151621
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
16161622
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
1617-
LOG.warn("Retrying.", e);
1618-
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1623+
if (i < maxAttempts) {
1624+
LOG.warn("Retrying.", e);
1625+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1626+
} else {
1627+
nodeSelector.getNextAliveNode(selectedEndpoint);
1628+
}
16191629
} else {
16201630
throw lastException;
16211631
}
16221632
}
16231633

1624-
if (i < retries) {
1634+
if (i < maxAttempts) {
16251635
try {
16261636
writer.onRetry();
16271637
} catch (IOException ioe) {
16281638
throw new ClientException("Failed to reset stream before next attempt", ioe);
16291639
}
16301640
}
16311641
}
1632-
String errMsg = requestExMsg("Insert", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId());
1642+
String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
16331643
LOG.warn(errMsg);
16341644
throw (lastException == null ? new ClientException(errMsg) : lastException);
16351645
};
@@ -1718,23 +1728,19 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
17181728
requestSettings.setQueryId(queryIdGenerator.get());
17191729
}
17201730

1731+
final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings());
1732+
final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1);
17211733
Supplier<QueryResponse> responseSupplier = () -> {
17221734
long startTime = System.nanoTime();
17231735
// Selecting some node
17241736
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
17251737
RuntimeException lastException = null;
1726-
for (int i = 0; i <= retries; i++) {
1738+
for (int i = 0; i <= maxAttempts; i++) {
17271739
ClassicHttpResponse httpResponse = null;
17281740
try {
1729-
boolean useMultipart = ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getOrDefault(requestSettings.getAllSettings());
1730-
if (queryParams != null && useMultipart) {
1731-
httpResponse = httpClientHelper.executeMultiPartRequest(selectedEndpoint,
1732-
requestSettings.getAllSettings(), sqlQuery);
1733-
} else {
1734-
httpResponse = httpClientHelper.executeRequest(selectedEndpoint,
1735-
requestSettings.getAllSettings(),
1736-
sqlQuery);
1737-
}
1741+
httpResponse = httpClientHelper.executeRequest(selectedEndpoint,
1742+
requestSettings.getAllSettings(),
1743+
sqlQuery);
17381744

17391745
OperationMetrics metrics = new OperationMetrics(clientStats);
17401746
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse
@@ -1758,14 +1764,18 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
17581764
String msg = requestExMsg("Query", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId());
17591765
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
17601766
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
1761-
LOG.warn("Retrying.", e);
1762-
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1767+
if (i < maxAttempts) {
1768+
LOG.warn("Retrying.", e);
1769+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
1770+
} else {
1771+
nodeSelector.getNextAliveNode(selectedEndpoint);
1772+
}
17631773
} else {
17641774
throw lastException;
17651775
}
17661776
}
17671777
}
1768-
String errMsg = requestExMsg("Query", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId());
1778+
String errMsg = requestExMsg("Query", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
17691779
LOG.warn(errMsg);
17701780
throw (lastException == null ? new ClientException(errMsg) : lastException);
17711781
};

‎client-v2/src/test/java/com/clickhouse/client/api/ClientFailoverUnitTest.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package com.clickhouse.client.api;
2+
3+
import com.clickhouse.client.api.Client;
4+
import com.clickhouse.client.api.query.QueryResponse;
5+
import com.github.tomakehurst.wiremock.WireMockServer;
6+
import com.github.tomakehurst.wiremock.client.WireMock;
7+
import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
8+
import org.testng.Assert;
9+
import org.testng.annotations.Test;
10+
11+
import java.util.concurrent.TimeUnit;
12+
13+
public class ClientFailoverUnitTest {
14+
15+
@Test
16+
public void testWireMockFailoverOnly() throws Exception {
17+
WireMockServer mockServer = new WireMockServer(WireMockConfiguration.options().dynamicPort());
18+
mockServer.start();
19+
try {
20+
mockServer.stubFor(WireMock.post(WireMock.anyUrl())
21+
.willReturn(WireMock.aResponse().withStatus(200)
22+
.withHeader("Content-Type", "text/plain")
23+
.withBody("")));
24+
25+
try (Client client = new Client.Builder()
26+
.addEndpoint("http://127.0.0.1:1") // dead endpoint
27+
.addEndpoint("http://localhost:" + mockServer.port()) // healthy mock endpoint
28+
.setUsername("default")
29+
.setPassword("password")
30+
.setDefaultDatabase("default")
31+
.setMaxRetries(3)
32+
.build()) {
33+
34+
try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) {
35+
Assert.assertNotNull(response);
36+
}
37+
}
38+
} finally {
39+
mockServer.stop();
40+
}
41+
}
42+
}

0 commit comments

Comments
 (0)