Skip to content

Commit 316bbac

Browse files
author
Paultagoras
committed
Adjusting retry mechanism
1 parent 5995a7e commit 316bbac

File tree

4 files changed

+94
-30
lines changed

4 files changed

+94
-30
lines changed

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

Lines changed: 15 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import java.lang.reflect.InvocationTargetException;
6464
import java.lang.reflect.Method;
6565
import java.net.ConnectException;
66+
import java.net.SocketTimeoutException;
6667
import java.net.URL;
6768
import java.nio.charset.StandardCharsets;
6869
import java.time.Duration;
@@ -1336,7 +1337,7 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13361337
// Selecting some node
13371338
ClickHouseNode selectedNode = getNextAliveNode();
13381339

1339-
ClientException lastException = null;
1340+
RuntimeException lastException = null;
13401341
for (int i = 0; i <= maxRetries; i++) {
13411342
// Execute request
13421343
try (ClassicHttpResponse httpResponse =
@@ -1376,19 +1377,17 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13761377
metrics.operationComplete();
13771378
metrics.setQueryId(queryId);
13781379
return new InsertResponse(metrics);
1379-
} catch (NoHttpResponseException | ConnectionRequestTimeoutException | ConnectTimeoutException | ConnectException e) {
1380-
lastException = httpClientHelper.wrapException("Insert request initiation failed", e);
1380+
} catch (Exception e) {
1381+
lastException = httpClientHelper.wrapException("Query request failed (Attempt " + (i + 1) + "/" + (maxRetries + 1) + ")", e);
13811382
if (httpClientHelper.shouldRetry(e, finalSettings.getAllSettings())) {
1382-
LOG.warn("Retrying", e);
1383+
LOG.warn("Retrying.", e);
13831384
selectedNode = getNextAliveNode();
13841385
} else {
13851386
throw lastException;
13861387
}
1387-
} catch (IOException e) {
1388-
throw new ClientException("Insert request failed", e);
13891388
}
13901389
}
1391-
throw new ClientException("Insert request failed after retries", lastException);
1390+
throw new ClientException("Insert request failed after attempts: " + (maxRetries + 1), lastException);
13921391
};
13931392

13941393
return runAsyncOperation(supplier, settings.getAllSettings());
@@ -1513,7 +1512,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15131512
// Selecting some node
15141513
ClickHouseNode selectedNode = getNextAliveNode();
15151514

1516-
ClientException lastException = null;
1515+
RuntimeException lastException = null;
15171516
for (int i = 0; i <= maxRetries; i++) {
15181517
// Execute request
15191518
try (ClassicHttpResponse httpResponse =
@@ -1538,16 +1537,14 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15381537
metrics.operationComplete();
15391538
metrics.setQueryId(queryId);
15401539
return new InsertResponse(metrics);
1541-
} catch (NoHttpResponseException | ConnectionRequestTimeoutException | ConnectTimeoutException | ConnectException e) {
1542-
lastException = httpClientHelper.wrapException("Insert request initiation failed", e);
1540+
} catch (Exception e) {
1541+
lastException = httpClientHelper.wrapException("Query request failed (Attempt " + (i + 1) + "/" + (maxRetries + 1) + ")", e);
15431542
if (httpClientHelper.shouldRetry(e, finalSettings.getAllSettings())) {
1544-
LOG.warn("Retrying", e);
1543+
LOG.warn("Retrying.", e);
15451544
selectedNode = getNextAliveNode();
15461545
} else {
15471546
throw lastException;
15481547
}
1549-
} catch (IOException e) {
1550-
throw new ClientException("Insert request failed", e);
15511548
}
15521549

15531550
if (i < maxRetries) {
@@ -1558,7 +1555,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15581555
}
15591556
}
15601557
}
1561-
throw new ClientException("Insert request failed after retries", lastException);
1558+
throw new ClientException("Insert request failed after attempts: " + (maxRetries + 1), lastException);
15621559
};
15631560
} else {
15641561
responseSupplier = () -> {
@@ -1672,7 +1669,7 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
16721669
responseSupplier = () -> {
16731670
// Selecting some node
16741671
ClickHouseNode selectedNode = getNextAliveNode();
1675-
ClientException lastException = null;
1672+
RuntimeException lastException = null;
16761673
for (int i = 0; i <= maxRetries; i++) {
16771674
try {
16781675
ClassicHttpResponse httpResponse =
@@ -1699,22 +1696,18 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
16991696

17001697
return new QueryResponse(httpResponse, finalSettings.getFormat(), finalSettings, metrics);
17011698

1702-
} catch (NoHttpResponseException | ConnectionRequestTimeoutException | ConnectTimeoutException | ConnectException e) {
1703-
lastException = httpClientHelper.wrapException("Query request initiation failed", e);
1699+
} catch (Exception e) {
1700+
lastException = httpClientHelper.wrapException("Query request failed (Attempt " + (i + 1) + "/" + (maxRetries + 1) + ")", e);
17041701
if (httpClientHelper.shouldRetry(e, finalSettings.getAllSettings())) {
17051702
LOG.warn("Retrying.", e);
17061703
selectedNode = getNextAliveNode();
17071704
} else {
17081705
throw lastException;
17091706
}
1710-
} catch (ClientException | ServerException e) {
1711-
throw e;
1712-
} catch (Exception e) {
1713-
throw new ClientException("Query request failed", e);
17141707
}
17151708
}
17161709

1717-
throw new ClientException("Query request failed after retries", lastException);
1710+
throw new ClientException("Query request failed after attempts: " + (maxRetries + 1), lastException);
17181711
};
17191712
} else {
17201713
ClickHouseRequest<?> request = oldClient.read(getServerNode());

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,4 +7,5 @@ public enum ClientFaultCause {
77
NoHttpResponse,
88
ConnectTimeout,
99
ConnectionRequestTimeout,
10+
SocketTimeout,
1011
}

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

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,7 @@
6060
import java.net.ConnectException;
6161
import java.net.InetSocketAddress;
6262
import java.net.NoRouteToHostException;
63+
import java.net.SocketTimeoutException;
6364
import java.net.URI;
6465
import java.net.URISyntaxException;
6566
import java.net.URL;
@@ -108,7 +109,7 @@ public HttpAPIClientHelper(Map<String, String> configuration) {
108109
boolean useHttpCompression = chConfiguration.getOrDefault("client.use_http_compression", "false").equalsIgnoreCase("true");
109110
LOG.info("client compression: {}, server compression: {}, http compression: {}", usingClientCompression, usingServerCompression, useHttpCompression);
110111

111-
defaultRetryCauses = SerializerUtils.parseEnumList(chConfiguration.get("client_retry_on_failures"), ClientFaultCause.class);
112+
defaultRetryCauses = SerializerUtils.parseEnumList(chConfiguration.get(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getKey()), ClientFaultCause.class);
112113
if (defaultRetryCauses.contains(ClientFaultCause.None)) {
113114
defaultRetryCauses.removeIf(c -> c != ClientFaultCause.None);
114115
}
@@ -398,7 +399,7 @@ public ClassicHttpResponse executeRequest(ClickHouseNode server, Map<String, Obj
398399
} catch (ConnectException | NoRouteToHostException e) {
399400
LOG.warn("Failed to connect to '{}': {}", server.getHost(), e.getMessage());
400401
throw new ClientException("Failed to connect", e);
401-
} catch (ConnectionRequestTimeoutException | ServerException | NoHttpResponseException | ClientException e) {
402+
} catch (ConnectionRequestTimeoutException | ServerException | NoHttpResponseException | ClientException | SocketTimeoutException e) {
402403
throw e;
403404
} catch (Exception e) {
404405
throw new ClientException("Failed to execute request", e);
@@ -576,32 +577,46 @@ public static <T> T getHeaderVal(Header header, T defaultValue, Function<String,
576577
return converter.apply(header.getValue());
577578
}
578579

579-
public boolean shouldRetry(Exception ex, Map<String, Object> requestSettings) {
580+
public boolean shouldRetry(Throwable ex, Map<String, Object> requestSettings) {
580581
Set<ClientFaultCause> retryCauses = (Set<ClientFaultCause>)
581-
requestSettings.getOrDefault("retry_on_failures", defaultRetryCauses);
582+
requestSettings.getOrDefault(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getKey(), defaultRetryCauses);
582583

583584
if (retryCauses.contains(ClientFaultCause.None)) {
584585
return false;
585586
}
586587

587-
if (ex instanceof NoHttpResponseException ) {
588+
if (ex instanceof NoHttpResponseException
589+
|| ex.getCause() instanceof NoHttpResponseException) {
588590
return retryCauses.contains(ClientFaultCause.NoHttpResponse);
589591
}
590592

591-
if (ex instanceof ConnectException || ex instanceof ConnectTimeoutException) {
593+
if (ex instanceof ConnectException
594+
|| ex instanceof ConnectTimeoutException
595+
|| ex.getCause() instanceof ConnectException
596+
|| ex.getCause() instanceof ConnectTimeoutException) {
592597
return retryCauses.contains(ClientFaultCause.ConnectTimeout);
593598
}
594599

595-
if (ex instanceof ConnectionRequestTimeoutException) {
600+
if (ex instanceof ConnectionRequestTimeoutException
601+
|| ex.getCause() instanceof ConnectionRequestTimeoutException) {
596602
return retryCauses.contains(ClientFaultCause.ConnectionRequestTimeout);
597603
}
598604

605+
if (ex instanceof SocketTimeoutException
606+
|| ex.getCause() instanceof SocketTimeoutException) {
607+
return retryCauses.contains(ClientFaultCause.SocketTimeout);
608+
}
609+
599610
return false;
600611
}
601612

602613
// This method wraps some client specific exceptions into specific ClientException or just ClientException
603614
// ClientException will be also wrapped
604-
public ClientException wrapException(String message, Exception cause) {
615+
public RuntimeException wrapException(String message, Exception cause) {
616+
if (cause instanceof ClientException || cause instanceof ServerException) {
617+
return (RuntimeException) cause;
618+
}
619+
605620
if (cause instanceof ConnectionRequestTimeoutException ||
606621
cause instanceof NoHttpResponseException ||
607622
cause instanceof ConnectTimeoutException ||

‎client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -606,6 +606,8 @@ public void testServerSettings() {
606606
} catch (Exception e) {
607607
e.printStackTrace();
608608
Assert.fail("Unexpected exception", e);
609+
} finally {
610+
mockServer.stop();
609611
}
610612
}
611613
}
@@ -1059,6 +1061,59 @@ public void testWithDefaultTimeouts() {
10591061
}
10601062
}
10611063

1064+
1065+
@Test(groups = { "integration" })
1066+
public void testTimeoutsWithRetry() {
1067+
if (isCloud()) {
1068+
return; // mocked server
1069+
}
1070+
1071+
WireMockServer faultyServer = new WireMockServer( WireMockConfiguration
1072+
.options().port(9090).notifier(new ConsoleNotifier(false)));
1073+
faultyServer.start();
1074+
1075+
// First request gets no response
1076+
faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
1077+
.inScenario("Timeout")
1078+
.withRequestBody(WireMock.containing("SELECT 1"))
1079+
.whenScenarioStateIs(STARTED)
1080+
.willSetStateTo("Failed")
1081+
.willReturn(WireMock.aResponse()
1082+
.withStatus(HttpStatus.SC_OK)
1083+
.withFixedDelay(5000)
1084+
.withHeader("X-ClickHouse-Summary",
1085+
"{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());
1086+
1087+
// Second request gets a response (retry)
1088+
faultyServer.addStubMapping(WireMock.post(WireMock.anyUrl())
1089+
.inScenario("Timeout")
1090+
.withRequestBody(WireMock.containing("SELECT 1"))
1091+
.whenScenarioStateIs("Failed")
1092+
.willSetStateTo("Done")
1093+
.willReturn(WireMock.aResponse()
1094+
.withStatus(HttpStatus.SC_OK)
1095+
.withFixedDelay(1000)
1096+
.withHeader("X-ClickHouse-Summary",
1097+
"{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());
1098+
1099+
try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", faultyServer.port(), false)
1100+
.setUsername("default")
1101+
.setPassword("")
1102+
.setSocketTimeout(3000)
1103+
.retryOnFailures(ClientFaultCause.SocketTimeout)
1104+
.build()) {
1105+
int startTime = (int) System.currentTimeMillis();
1106+
try {
1107+
client.query("SELECT 1").get();
1108+
} catch (Exception e) {
1109+
Assert.fail("Elapsed Time: " + (System.currentTimeMillis() - startTime), e);
1110+
}
1111+
} finally {
1112+
faultyServer.stop();
1113+
}
1114+
}
1115+
1116+
10621117
protected Client.Builder newClient() {
10631118
ClickHouseNode node = getServer(ClickHouseProtocol.HTTP);
10641119
boolean isSecure = isCloud();

0 commit comments

Comments
 (0)