Skip to content

Commit 96934a9

Browse files
kishansinghifs1kishansinghifs1
committed
fix: throw ConnectException on 502/503 errors, add quarantine validation, and refactor node selection logic in Client
1 parent 790cdeb commit 96934a9

6 files changed

Lines changed: 104 additions & 63 deletions

File tree

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

Lines changed: 7 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -1317,7 +1317,7 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13171317
Supplier<InsertResponse> supplier = () -> {
13181318
long startTime = System.nanoTime();
13191319
// Selecting some node
1320-
Endpoint selectedEndpoint = getEndpoint();
1320+
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
13211321

13221322
RuntimeException lastException = null;
13231323
for (int i = 0; i <= maxRetries; i++) {
@@ -1343,14 +1343,6 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13431343
out.close();
13441344
})) {
13451345

1346-
1347-
// Check response
1348-
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
1349-
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
1350-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1351-
continue;
1352-
}
1353-
13541346
ClientStatisticsHolder clientStats = globalClientStats.remove(operationId);
13551347
OperationMetrics metrics = new OperationMetrics(clientStats);
13561348
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}");
@@ -1364,7 +1356,7 @@ public CompletableFuture<InsertResponse> insert(String tableName, List<?> data,
13641356
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
13651357
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
13661358
LOG.warn("Retrying.", e);
1367-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1359+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
13681360
} else {
13691361
throw lastException;
13701362
}
@@ -1539,7 +1531,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15391531
responseSupplier = () -> {
15401532
long startTime = System.nanoTime();
15411533
// Selecting some node
1542-
Endpoint selectedEndpoint = getEndpoint();
1534+
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
15431535

15441536
RuntimeException lastException = null;
15451537
for (int i = 0; i <= retries; i++) {
@@ -1551,21 +1543,6 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15511543
out.close();
15521544
})) {
15531545

1554-
1555-
// Check response
1556-
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
1557-
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
1558-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1559-
if (i < retries) {
1560-
try {
1561-
writer.onRetry();
1562-
} catch (IOException ioe) {
1563-
throw new ClientException("Failed to reset stream before next attempt", ioe);
1564-
}
1565-
}
1566-
continue;
1567-
}
1568-
15691546
OperationMetrics metrics = new OperationMetrics(finalClientStats);
15701547
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}");
15711548
ProcessParser.parseSummary(summary, metrics);
@@ -1578,7 +1555,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
15781555
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
15791556
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
15801557
LOG.warn("Retrying.", e);
1581-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1558+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
15821559
} else {
15831560
throw lastException;
15841561
}
@@ -1675,27 +1652,20 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
16751652
Supplier<QueryResponse> responseSupplier = () -> {
16761653
long startTime = System.nanoTime();
16771654
// Selecting some node
1678-
Endpoint selectedEndpoint = getEndpoint();
1655+
Endpoint selectedEndpoint = nodeSelector.getEndpoint();
16791656
RuntimeException lastException = null;
16801657
for (int i = 0; i <= retries; i++) {
16811658
ClassicHttpResponse httpResponse = null;
16821659
try {
16831660
boolean useMultipart = ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getOrDefault(requestSettings.getAllSettings());
16841661
if (queryParams != null && useMultipart) {
16851662
httpResponse = httpClientHelper.executeMultiPartRequest(selectedEndpoint,
1686-
requestSettings.getAllSettings(), sqlQuery);
1663+
requestSettings.getAllSettings(), sqlQuery);
16871664
} else {
16881665
httpResponse = httpClientHelper.executeRequest(selectedEndpoint,
16891666
requestSettings.getAllSettings(),
16901667
sqlQuery);
16911668
}
1692-
// Check response
1693-
if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
1694-
LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime));
1695-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1696-
HttpAPIClientHelper.closeQuietly(httpResponse);
1697-
continue;
1698-
}
16991669

17001670
OperationMetrics metrics = new OperationMetrics(clientStats);
17011671
String summary = HttpAPIClientHelper.getHeaderVal(httpResponse
@@ -1720,7 +1690,7 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
17201690
lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId());
17211691
if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings())) {
17221692
LOG.warn("Retrying.", e);
1723-
selectedEndpoint = getNextAliveNode(selectedEndpoint);
1693+
selectedEndpoint = nodeSelector.getNextAliveNode(selectedEndpoint);
17241694
} else {
17251695
throw lastException;
17261696
}
@@ -2232,24 +2202,7 @@ public void updateAccessToken(String accessToken) {
22322202
this.credentialsManager.setAccessToken(accessToken);
22332203
}
22342204

2235-
/**
2236-
* Returns the first alive endpoint for the initial request attempt.
2237-
* Keeps affinity to the primary server when it is healthy.
2238-
*/
2239-
private Endpoint getEndpoint() {
2240-
return nodeSelector.getEndpoint();
2241-
}
22422205

2243-
/**
2244-
* Quarantines the failed endpoint and returns the next alive endpoint.
2245-
* Used inside retry loops after a retryable failure.
2246-
*
2247-
* @param failedEndpoint the endpoint that just failed
2248-
* @return the next alive endpoint
2249-
*/
2250-
private Endpoint getNextAliveNode(Endpoint failedEndpoint) {
2251-
return nodeSelector.getNextAliveNode(failedEndpoint);
2252-
}
22532206

22542207
public static final String VALUES_LIST_DELIMITER = ",";
22552208

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

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,9 +54,6 @@ public String getQueryId() {
5454
}
5555

5656
private boolean discoverIsRetryable(int code, String message, int transportProtocolCode) {
57-
if (transportProtocolCode == 503) {
58-
return true;
59-
}
6057
// Let's check if we have a ServerException to reference the error code
6158
// https://github.com/ClickHouse/ClickHouse/blob/master/src/Common/ErrorCodes.cpp
6259
switch (code) { // UNEXPECTED_END_OF_FILE

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -577,9 +577,10 @@ private ClassicHttpResponse doPostRequest(Map<String, Object> requestConfig, Htt
577577
throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings.");
578578
} else if (httpResponse.getCode() == HttpStatus.SC_BAD_GATEWAY) {
579579
httpResponse.close();
580-
throw new ClientException("Server returned '502 Bad gateway'. Check network and proxy settings.");
580+
throw new ConnectException("Server returned '502 Bad gateway'. Check network and proxy settings.");
581581
} else if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) {
582-
return httpResponse;
582+
httpResponse.close();
583+
throw new ConnectException("Server returned '503 Service unavailable'.");
583584
} else if (httpResponse.getCode() >= HttpStatus.SC_BAD_REQUEST || httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) {
584585
try {
585586
throw readError(req, httpResponse);

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,9 @@ Endpoint getEndpoint() {
2121
}
2222

2323
void markFailed(long quarantineMs) {
24+
if (quarantineMs <= 0) {
25+
throw new IllegalArgumentException("Quarantine duration must be positive: " + quarantineMs);
26+
}
2427
this.failedUntil = System.currentTimeMillis() + quarantineMs;
2528
}
2629

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

Lines changed: 77 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,78 @@
1-
package com.clickhouse.client.api.internal;
2-
3-
public class HttpAPIClientHelperTest {
4-
1+
package com.clickhouse.client.api.internal;
2+
3+
import com.clickhouse.client.api.transport.Endpoint;
4+
import com.clickhouse.client.api.transport.HttpEndpoint;
5+
import net.jpountz.lz4.LZ4Factory;
6+
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
7+
import org.apache.hc.core5.http.ClassicHttpResponse;
8+
import org.apache.hc.core5.http.HttpEntity;
9+
import org.testng.Assert;
10+
import org.testng.annotations.Test;
11+
12+
import java.lang.reflect.Field;
13+
import java.net.ConnectException;
14+
import java.util.HashMap;
15+
import java.util.Map;
16+
17+
import static org.mockito.ArgumentMatchers.any;
18+
import static org.mockito.Mockito.mock;
19+
import static org.mockito.Mockito.when;
20+
21+
public class HttpAPIClientHelperTest {
22+
23+
@Test
24+
public void testExecuteRequestThrowsConnectExceptionOn502() throws Exception {
25+
Map<String, Object> configuration = new HashMap<>();
26+
HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance());
27+
28+
CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class);
29+
Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient");
30+
httpClientField.setAccessible(true);
31+
httpClientField.set(helper, mockHttpClient);
32+
33+
ClassicHttpResponse mockResponse = mock(ClassicHttpResponse.class);
34+
when(mockResponse.getCode()).thenReturn(502);
35+
HttpEntity mockEntity = mock(HttpEntity.class);
36+
when(mockResponse.getEntity()).thenReturn(mockEntity);
37+
38+
when(mockHttpClient.executeOpen(any(), any(), any())).thenReturn(mockResponse);
39+
40+
Endpoint endpoint = new HttpEndpoint("localhost", 8123, false, "/");
41+
try {
42+
helper.executeRequest(endpoint, new HashMap<>(), "SELECT 1");
43+
Assert.fail("Expected ConnectException to be thrown");
44+
} catch (ConnectException e) {
45+
// expected
46+
} catch (Exception e) {
47+
Assert.fail("Expected ConnectException to be thrown, but got: " + e.getClass().getName(), e);
48+
}
49+
}
50+
51+
@Test
52+
public void testExecuteRequestThrowsConnectExceptionOn503() throws Exception {
53+
Map<String, Object> configuration = new HashMap<>();
54+
HttpAPIClientHelper helper = new HttpAPIClientHelper(configuration, null, false, LZ4Factory.fastestInstance());
55+
56+
CloseableHttpClient mockHttpClient = mock(CloseableHttpClient.class);
57+
Field httpClientField = HttpAPIClientHelper.class.getDeclaredField("httpClient");
58+
httpClientField.setAccessible(true);
59+
httpClientField.set(helper, mockHttpClient);
60+
61+
ClassicHttpResponse mockResponse = mock(ClassicHttpResponse.class);
62+
when(mockResponse.getCode()).thenReturn(503);
63+
HttpEntity mockEntity = mock(HttpEntity.class);
64+
when(mockResponse.getEntity()).thenReturn(mockEntity);
65+
66+
when(mockHttpClient.executeOpen(any(), any(), any())).thenReturn(mockResponse);
67+
68+
Endpoint endpoint = new HttpEndpoint("localhost", 8123, false, "/");
69+
try {
70+
helper.executeRequest(endpoint, new HashMap<>(), "SELECT 1");
71+
Assert.fail("Expected ConnectException to be thrown");
72+
} catch (ConnectException e) {
73+
// expected
74+
} catch (Exception e) {
75+
Assert.fail("Expected ConnectException to be thrown, but got: " + e.getClass().getName(), e);
76+
}
77+
}
578
}

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,20 @@ public void testEndpointStateQuarantineAndExpiry() throws InterruptedException {
2929
Assert.assertTrue(state.isAlive(), "Endpoint state should be alive again after quarantine expires");
3030
}
3131

32+
@Test(expectedExceptions = IllegalArgumentException.class)
33+
public void testEndpointStateQuarantineValidation() {
34+
Endpoint ep = new HttpEndpoint("localhost", 8123, false, "/");
35+
EndpointState state = new EndpointState(ep);
36+
state.markFailed(0);
37+
}
38+
39+
@Test(expectedExceptions = IllegalArgumentException.class)
40+
public void testEndpointStateQuarantineValidationNegative() {
41+
Endpoint ep = new HttpEndpoint("localhost", 8123, false, "/");
42+
EndpointState state = new EndpointState(ep);
43+
state.markFailed(-10);
44+
}
45+
3246
@Test
3347
public void testClientNodeSelectorAffinityAndQuarantine() {
3448
Endpoint epA = new HttpEndpoint("localhost", 8123, false, "/");

0 commit comments

Comments
 (0)