From 63633b582c4e70cefd165221f161f519b6cf8f7f Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 19:59:17 +0000 Subject: [PATCH 1/6] fix(client-v2): keep cancellation per operation so it is not lost between retry attempts Cancellation state lived on the per-attempt TransportRequest, so a cancelTransportRequest(queryId) landing between two attempts of a retried operation was silently dropped: on the query path the next attempt overwrote the registry entry with a fresh un-cancelled request, and on both insert paths the entry was unregistered after every attempt, so the cancel was a complete no-op. The retry guard then saw an un-cancelled (or missing) request and the operation retried and could succeed. Cancellation is now tracked per operation: the registry holds an OngoingOperation with a sticky cancelled flag, registration lives for the whole operation on all three retry loops, a request attached to an already cancelled operation is cancelled right away, and every retry iteration re-checks the flag before issuing another request. A cancelled operation now fails with the same TransportException("Request was cancelled on client side") that an in-flight cancellation produces. Fixes: https://github.com/ClickHouse/clickhouse-java/issues/2989 --- CHANGELOG.md | 5 + .../com/clickhouse/client/api/Client.java | 186 ++++++++++++------ .../api/transport/TransportBaseTests.java | 96 +++++++++ docs/features.md | 2 +- 4 files changed, 227 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 52c92a57a..3cd77ae32 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,11 @@ ### Bug Fixes +- **[client-v2]** Fixed `Client.cancelTransportRequest(queryId)` being silently dropped when it landed between two + attempts of a retried operation (query, POJO insert and stream insert): the operation issued the next attempt anyway + and could complete successfully. Cancellation is now tracked per operation instead of per transport request, so a + cancelled operation stays cancelled for every following attempt and the retry loop stops instead of sending another + request. (https://github.com/ClickHouse/clickhouse-java/issues/2989) - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial read. (https://github.com/ClickHouse/clickhouse-java/issues/2985) 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 3d61576ac..035a11ac8 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 @@ -144,7 +144,7 @@ public class Client implements AutoCloseable { private final Map> typeHintMapping; - private final ConcurrentHashMap ongoingRequests = new ConcurrentHashMap<>(REQ_REGISTRY_SIZE); + private final ConcurrentHashMap ongoingRequests = new ConcurrentHashMap<>(REQ_REGISTRY_SIZE); // Server context private String dbUser; @@ -1465,50 +1465,60 @@ public CompletableFuture insert(String tableName, List data, Endpoint selectedEndpoint = nodeSelector.getEndpoint(); final String queryId = requestSettings.getQueryId(); RuntimeException lastException = null; - for (int i = 0; i <= maxAttempts; i++) { - // Execute request - TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), - out -> { - out.write("INSERT INTO ".getBytes()); - out.write(tableName.getBytes()); - out.write(" \n FORMAT ".getBytes()); - out.write(format.name().getBytes()); - out.write(" \n".getBytes()); - for (Object obj : data) { - - for (POJOFieldSerializer serializer : serializersForTable) { - try { - serializer.serialize(obj, out); - } catch (InvocationTargetException | IllegalAccessException e) { - throw new DataSerializationException(obj, serializer, e); + final OngoingOperation ongoingOperation = registerOperation(queryId); + try { + for (int i = 0; i <= maxAttempts; i++) { + // Cancellation belongs to the operation, so one that was cancelled between two + // attempts must not issue another request. + if (i > 0 && !requestIsNotCancelled(ongoingOperation)) { + throw cancelledException(lastException, queryId); + } + // Execute request + TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), + out -> { + out.write("INSERT INTO ".getBytes()); + out.write(tableName.getBytes()); + out.write(" \n FORMAT ".getBytes()); + out.write(format.name().getBytes()); + out.write(" \n".getBytes()); + for (Object obj : data) { + + for (POJOFieldSerializer serializer : serializersForTable) { + try { + serializer.serialize(obj, out); + } catch (InvocationTargetException | IllegalAccessException e) { + throw new DataSerializationException(obj, serializer, e); + } } } - } - out.close(); - }); + out.close(); + }); - registerTransportReq(queryId, transportRequest); + registerTransportReq(ongoingOperation, transportRequest); - try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { - ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); - OperationMetrics metrics = completeOperation(transportResponse, clientStats, requestSettings.getQueryId()); + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { + ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); + OperationMetrics metrics = completeOperation(transportResponse, clientStats, requestSettings.getQueryId()); - return new InsertResponse(transportResponse, metrics); - } catch (Exception e) { - String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); - lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(queryId)) { - if (i < maxAttempts) { - selectedEndpoint = logRetryAndSelectNextNode("Insert", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); + return new InsertResponse(transportResponse, metrics); + } catch (Exception e) { + String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); + lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(ongoingOperation)) { + if (i < maxAttempts) { + selectedEndpoint = logRetryAndSelectNextNode("Insert", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); + } else { + nodeSelector.getNextAliveNode(selectedEndpoint); + } } else { - nodeSelector.getNextAliveNode(selectedEndpoint); + throw lastException; } - } else { - throw lastException; } - } finally { - unregisterTransportReq(queryId); } + } finally { + // The operation is over: it will not issue another request, so its cancellation state + // is no longer needed. + unregisterOperation(queryId, ongoingOperation); } String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1686,15 +1696,21 @@ public CompletableFuture insert(String tableName, RuntimeException lastException = null; final String queryId = requestSettings.getQueryId(); + final OngoingOperation ongoingOperation = registerOperation(queryId); try { for (int i = 0; i <= maxAttempts; i++) { + // Cancellation belongs to the operation, so one that was cancelled between two + // attempts (for instance from DataStreamWriter#onRetry()) must not issue another request. + if (i > 0 && !requestIsNotCancelled(ongoingOperation)) { + throw cancelledException(lastException, queryId); + } // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), out -> { writer.onOutput(out); out.close(); }); - registerTransportReq(queryId, transportRequest); + registerTransportReq(ongoingOperation, transportRequest); try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { OperationMetrics metrics = completeOperation(transportResponse, finalClientStats, requestSettings.getQueryId()); @@ -1702,7 +1718,7 @@ public CompletableFuture insert(String tableName, } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(requestSettings.getQueryId())) { + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(ongoingOperation)) { if (i < maxAttempts) { selectedEndpoint = logRetryAndSelectNextNode("Insert (stream)", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); } else { @@ -1711,10 +1727,6 @@ public CompletableFuture insert(String tableName, } else { throw lastException; } - } finally { - // Insert completes once the request returns; the response exposes no stream to read afterwards, - // so the request is no longer cancellable and can be unregistered. - unregisterTransportReq(requestSettings.getQueryId()); } if (i < maxAttempts) { @@ -1726,7 +1738,7 @@ public CompletableFuture insert(String tableName, } } } finally { - unregisterTransportReq(queryId); + unregisterOperation(queryId, ongoingOperation); } String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1826,10 +1838,16 @@ public CompletableFuture query(String sqlQuery, Map 0 && !requestIsNotCancelled(ongoingOperation)) { + throw cancelledException(lastException, queryId); + } TransportRequest request = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), sqlQuery); - registerTransportReq(queryId, request); + registerTransportReq(ongoingOperation, request); TransportResponse transportResp = null; try { transportResp = httpClientHelper.executeRequest(request); @@ -1845,7 +1863,7 @@ public CompletableFuture query(String sqlQuery, Map query(String sqlQuery, Map query(String sqlQuery, Map queryParams) { @@ -2416,15 +2479,16 @@ public void updateAccessToken(String accessToken) { * Tries to cancel ongoing request. This method cancels IO operations but doesn't * kill query on server side. Original queryId should be used to cancel the request. * This operation cancels only operations on client side and only that still waiting - * for response. + * for response. Cancellation applies to the whole operation: a retrying operation stops + * instead of issuing another attempt. * * @param queryId - original query id that was passed in operation settings. */ public void cancelTransportRequest(String queryId) { Objects.requireNonNull(queryId, "queryId should be not null"); - TransportRequest req = ongoingRequests.get(queryId); - if (req != null) { - req.cancel(); + OngoingOperation operation = ongoingRequests.get(queryId); + if (operation != null) { + operation.cancel(); } } diff --git a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java index 19dfe0bec..03f3c8f9c 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -7,7 +7,9 @@ import com.clickhouse.client.ClickHouseServerForTest; import com.clickhouse.client.api.Client; import com.clickhouse.client.api.ClientFaultCause; +import com.clickhouse.client.api.DataStreamWriter; import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.TransportException; import com.clickhouse.client.api.enums.Protocol; import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; @@ -28,7 +30,9 @@ import org.testng.annotations.Test; import java.io.ByteArrayInputStream; +import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.List; @@ -521,6 +525,98 @@ public static Object[][] cancelDuringRetryProvider() { }; } + /** + * Cancellation belongs to the operation, not to a single transport request: a cancel that lands + * between two attempts of a retried operation must stop it, even though the request of the failed + * attempt is already completed. {@link DataStreamWriter#onRetry()} is called by the client exactly + * in that window, so it cancels deterministically. Cancellation does not outlive the operation, so the + * same query id still works afterwards. A cancel for an unrelated query id must leave the operation + * alone and let it recover on the retry. + */ + @Test(groups = {"integration"}, dataProvider = "cancelBetweenAttemptsProvider") + public void testCancelBetweenRetryAttempts(String name, boolean cancelOwnQueryId) throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("CancelBetweenAttempts") + .whenScenarioStateIs(STARTED) + .willSetStateTo("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(RETRYABLE_CODE)) + .withBody(RETRYABLE_BODY)).build()); + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("CancelBetweenAttempts") + .whenScenarioStateIs("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + + String queryId = "cancel-between-attempts-" + UUID.randomUUID(); + String cancelledQueryId = cancelOwnQueryId ? queryId : queryId + "-other"; + AtomicInteger retryCallbacks = new AtomicInteger(); + + try (Client client = mockServerClient(mockServer, 3)) { + DataStreamWriter writer = new DataStreamWriter() { + @Override + public void onOutput(OutputStream out) throws IOException { + out.write("1\n".getBytes(StandardCharsets.US_ASCII)); + } + + @Override + public void onRetry() { + retryCallbacks.incrementAndGet(); + client.cancelTransportRequest(cancelledQueryId); + } + }; + + Throwable opError = null; + try (InsertResponse response = client.insert("table01", writer, ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } catch (Exception e) { + opError = e; + } + + Assert.assertEquals(retryCallbacks.get(), 1, + "[" + name + "] the first attempt must fail with a retryable error"); + int attempts = mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(); + if (cancelOwnQueryId) { + Assert.assertNotNull(opError, "[" + name + "] a cancelled operation must fail"); + Assert.assertTrue(opError instanceof TransportException + || opError.getCause() instanceof TransportException, + "[" + name + "] a cancelled operation must fail as cancelled, was: " + opError); + Assert.assertEquals(attempts, 1, + "[" + name + "] a cancelled operation must not issue another attempt"); + + // Cancellation is bound to the operation and does not outlive it: the same query id can be + // used again by the next operation. + try (InsertResponse response = client.insert("table01", writer, ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } else { + Assert.assertNull(opError, "[" + name + "] an operation that was not cancelled must recover: " + opError); + Assert.assertEquals(attempts, 2, + "[" + name + "] an operation that was not cancelled must be retried"); + } + } finally { + mockServer.stop(); + } + } + + @DataProvider(name = "cancelBetweenAttemptsProvider") + public static Object[][] cancelBetweenAttemptsProvider() { + return new Object[][]{ + {"cancelled", true}, + {"other-query-id", false} + }; + } + /** * Reissues {@link Client#cancelTransportRequest(String)} until the worker stops or the timeout elapses. * Retrying makes the test robust against thread-scheduling races where a single cancel could land before diff --git a/docs/features.md b/docs/features.md index 915e8ac9e..dc50588fe 100644 --- a/docs/features.md +++ b/docs/features.md @@ -33,7 +33,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Server information loading: Can refresh server version, current user, and server time zone information. - Compression support: Supports response compression, ClickHouse LZ4 request/response compression, HTTP content compression, and caller-supplied precompressed insert bodies. - Retry behavior: Can retry failed operations for configured failure causes and retry limits. -- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. +- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. Cancellation applies to the whole operation rather than to a single attempt: a cancelled operation that is being retried stops instead of issuing another request, including when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`). - Metrics and observability: Exposes client/server operation metrics and optionally integrates connection-pool gauges with Micrometer. - Configuration surface: Supports arbitrary client options, cookies, custom headers, server-setting prefixes, client naming, query id suppliers, and buffer sizing. - SQL helpers: Includes SQL quoting and temporal formatting helpers used by callers building SQL text safely. From f4ef787997e8c440cae0f1c4869e7d5e362c93f4 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:21:10 +0000 Subject: [PATCH 2/6] fix(client-v2): register an operation for cancellation when it starts, not when its first request is created With useAsyncRequests(true) the operation body runs on the shared executor, so an operation registered itself in the cancellation registry only once the executor picked it up. A cancelTransportRequest(queryId) issued right after the call returned found no entry, was silently dropped, and the operation then ran to completion - the front edge of the same window that was closed between attempts. Operations are now registered on the calling thread, before being submitted, and the cancellation guard runs before every attempt (including the first), so a cancelled operation fails with TransportException("Request was cancelled on client side") without sending a request. A registration is dropped again when the submission is rejected, and the registry is cleared on close() because an operation still queued when the executor is shut down never runs. Fixes: https://github.com/ClickHouse/clickhouse-java/issues/2989 --- CHANGELOG.md | 5 +- .../com/clickhouse/client/api/Client.java | 55 +++++++---- .../api/transport/TransportBaseTests.java | 98 +++++++++++++++++++ docs/features.md | 2 +- 4 files changed, 141 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3cd77ae32..b3598c8ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,7 +48,10 @@ attempts of a retried operation (query, POJO insert and stream insert): the operation issued the next attempt anyway and could complete successfully. Cancellation is now tracked per operation instead of per transport request, so a cancelled operation stays cancelled for every following attempt and the retry loop stops instead of sending another - request. (https://github.com/ClickHouse/clickhouse-java/issues/2989) + request. A cancellation that lands before the operation sent its first request is no longer dropped either - it was + possible to lose one with `useAsyncRequests(true)`, where the operation body runs on the shared executor: an operation + is now registered for cancellation when it is started rather than when its first request is created, so it stops + before any request reaches the server. (https://github.com/ClickHouse/clickhouse-java/issues/2989) - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial read. (https://github.com/ClickHouse/clickhouse-java/issues/2985) 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 035a11ac8..a997d0b00 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 @@ -266,6 +266,10 @@ public void close() { LOG.debug("Skip closing operation executor because not owned by client"); } + // An operation is registered when it is started, so one that is still queued when the executor is + // shut down never runs and never removes its own registration. + ongoingRequests.clear(); + if (httpClientHelper != null) { httpClientHelper.close(); } @@ -1459,18 +1463,18 @@ public CompletableFuture insert(String tableName, List data, if (requestSettings.getQueryId() == null && queryIdGenerator != null) { requestSettings.setQueryId(queryIdGenerator.get()); } + final String queryId = requestSettings.getQueryId(); + final OngoingOperation ongoingOperation = registerOperation(queryId); Supplier supplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); - final String queryId = requestSettings.getQueryId(); RuntimeException lastException = null; - final OngoingOperation ongoingOperation = registerOperation(queryId); try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled between two + // Cancellation belongs to the operation, so one that was cancelled before or between // attempts must not issue another request. - if (i > 0 && !requestIsNotCancelled(ongoingOperation)) { + if (!requestIsNotCancelled(ongoingOperation)) { throw cancelledException(lastException, queryId); } // Execute request @@ -1526,7 +1530,7 @@ public CompletableFuture insert(String tableName, List data, throw (lastException == null ? new ClientException(errMsg) : lastException); }; - return runAsyncOperation(supplier, requestSettings.getAllSettings()); + return runAsyncOperation(supplier, requestSettings.getAllSettings(), queryId, ongoingOperation); } /** @@ -1689,19 +1693,19 @@ public CompletableFuture insert(String tableName, final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings()); final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1); + final String queryId = requestSettings.getQueryId(); + final OngoingOperation ongoingOperation = registerOperation(queryId); Supplier responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); RuntimeException lastException = null; - final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId); try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled between two + // Cancellation belongs to the operation, so one that was cancelled before or between // attempts (for instance from DataStreamWriter#onRetry()) must not issue another request. - if (i > 0 && !requestIsNotCancelled(ongoingOperation)) { + if (!requestIsNotCancelled(ongoingOperation)) { throw cancelledException(lastException, queryId); } // Execute request @@ -1746,7 +1750,7 @@ public CompletableFuture insert(String tableName, throw (lastException == null ? new ClientException(errMsg) : lastException); }; - return runAsyncOperation(responseSupplier, requestSettings.getAllSettings()); + return runAsyncOperation(responseSupplier, requestSettings.getAllSettings(), queryId, ongoingOperation); } /** @@ -1832,18 +1836,18 @@ public CompletableFuture query(String sqlQuery, Map responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); RuntimeException lastException = null; - final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId); try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled between two + // Cancellation belongs to the operation, so one that was cancelled before or between // attempts must not issue another request. - if (i > 0 && !requestIsNotCancelled(ongoingOperation)) { + if (!requestIsNotCancelled(ongoingOperation)) { throw cancelledException(lastException, queryId); } TransportRequest request = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), sqlQuery); @@ -1883,7 +1887,7 @@ public CompletableFuture query(String sqlQuery, Map CompletableFuture runAsyncOperation(Supplier resultSupplier, Map requestSettings, + String queryId, OngoingOperation operation) { + try { + return runAsyncOperation(resultSupplier, requestSettings); + } catch (RuntimeException | Error e) { + unregisterOperation(queryId, operation); + throw e; + } + } + private CompletableFuture runAsyncOperation(Supplier resultSupplier, Map requestSettings) { boolean isAsync = MapUtils.getFlag(requestSettings, configuration, ClientConfigProperties.ASYNC_OPERATIONS.getKey()); if (isAsync) { @@ -2479,8 +2499,9 @@ public void updateAccessToken(String accessToken) { * Tries to cancel ongoing request. This method cancels IO operations but doesn't * kill query on server side. Original queryId should be used to cancel the request. * This operation cancels only operations on client side and only that still waiting - * for response. Cancellation applies to the whole operation: a retrying operation stops - * instead of issuing another attempt. + * for response. Cancellation applies to the whole operation: it is effective as soon as the + * operation was started (so also for an asynchronous operation that has not sent its first + * request yet) and a retrying operation stops instead of issuing another attempt. * * @param queryId - original query id that was passed in operation settings. */ diff --git a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java index 03f3c8f9c..5aac71df8 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -37,6 +37,10 @@ import java.util.Collections; import java.util.List; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -617,6 +621,100 @@ public static Object[][] cancelBetweenAttemptsProvider() { }; } + /** + * An asynchronous operation runs its body on the shared executor, so a cancel issued right after the + * call returned can land before the operation has sent its first request. An operation is registered for + * cancellation when it is started, not when its first request is created, so such a cancel must stop it + * before any request reaches the server instead of being silently dropped. A cancel for an unrelated + * query id must leave the queued operation alone. + */ + @Test(groups = {"integration"}, dataProvider = "cancelBeforeFirstAttemptProvider") + public void testCancelBeforeFirstAttempt(String name, boolean insert, boolean cancelOwnQueryId) throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + + ExecutorService operationExecutor = Executors.newSingleThreadExecutor(); + CountDownLatch executorBusy = new CountDownLatch(1); + CountDownLatch releaseExecutor = new CountDownLatch(1); + String queryId = "cancel-before-first-attempt-" + UUID.randomUUID(); + String cancelledQueryId = cancelOwnQueryId ? queryId : queryId + "-other"; + + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressClientRequest(false) + .compressServerResponse(false) + .useAsyncRequests(true) + .setSharedOperationExecutor(operationExecutor) + .build()) { + + // Occupies the only worker thread so the operation stays queued: it has not created its first + // request yet when the cancel is issued. + operationExecutor.submit(() -> { + executorBusy.countDown(); + return releaseExecutor.await(30, TimeUnit.SECONDS); + }); + Assert.assertTrue(executorBusy.await(30, TimeUnit.SECONDS), + "[" + name + "] operation executor did not start"); + + CompletableFuture operation = insert + ? client.insert("table01", out -> out.write("1\n".getBytes(StandardCharsets.US_ASCII)), + ClickHouseFormat.TSV, new InsertSettings().setQueryId(queryId)) + : client.query("SELECT 1", new QuerySettings().setQueryId(queryId)); + client.cancelTransportRequest(cancelledQueryId); + releaseExecutor.countDown(); + + Throwable opError = null; + try { + Object response = operation.get(30, TimeUnit.SECONDS); + Assert.assertNotNull(response); + ((AutoCloseable) response).close(); + } catch (Exception e) { + opError = e; + } + + int requests = mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(); + if (cancelOwnQueryId) { + Assert.assertNotNull(opError, "[" + name + "] a cancelled operation must fail"); + Throwable cancellation = opError instanceof TransportException ? opError : opError.getCause(); + Assert.assertTrue(cancellation instanceof TransportException, + "[" + name + "] a cancelled operation must fail as cancelled, was: " + opError); + Assert.assertTrue(cancellation.getMessage().contains("cancelled on client side"), + "[" + name + "] a cancelled operation must fail as cancelled, was: " + cancellation.getMessage()); + Assert.assertEquals(requests, 0, + "[" + name + "] a cancelled operation must not send a request"); + } else { + Assert.assertNull(opError, + "[" + name + "] an operation that was not cancelled must complete: " + opError); + Assert.assertEquals(requests, 1, + "[" + name + "] an operation that was not cancelled must send its request"); + } + } finally { + releaseExecutor.countDown(); + operationExecutor.shutdownNow(); + mockServer.stop(); + } + } + + @DataProvider(name = "cancelBeforeFirstAttemptProvider") + public static Object[][] cancelBeforeFirstAttemptProvider() { + return new Object[][]{ + {"query-cancelled", false, true}, + {"query-other-query-id", false, false}, + {"insert-cancelled", true, true}, + {"insert-other-query-id", true, false} + }; + } + /** * Reissues {@link Client#cancelTransportRequest(String)} until the worker stops or the timeout elapses. * Retrying makes the test robust against thread-scheduling races where a single cancel could land before diff --git a/docs/features.md b/docs/features.md index dc50588fe..7cedab081 100644 --- a/docs/features.md +++ b/docs/features.md @@ -33,7 +33,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Server information loading: Can refresh server version, current user, and server time zone information. - Compression support: Supports response compression, ClickHouse LZ4 request/response compression, HTTP content compression, and caller-supplied precompressed insert bodies. - Retry behavior: Can retry failed operations for configured failure causes and retry limits. -- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. Cancellation applies to the whole operation rather than to a single attempt: a cancelled operation that is being retried stops instead of issuing another request, including when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`). +- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. Cancellation applies to the whole operation rather than to a single attempt: a cancelled operation that is being retried stops instead of issuing another request, including when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`) or before the first request was sent (an asynchronous operation runs its body on the shared executor, so a cancel issued right after the call returned can arrive first). - Metrics and observability: Exposes client/server operation metrics and optionally integrates connection-pool gauges with Micrometer. - Configuration surface: Supports arbitrary client options, cookies, custom headers, server-setting prefixes, client naming, query id suppliers, and buffer sizing. - SQL helpers: Includes SQL quoting and temporal formatting helpers used by callers building SQL text safely. From dd3aab1493c53ec089b04b9c0b478b0ecab51bc9 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 20:45:27 +0000 Subject: [PATCH 3/6] refactor(client-v2): handle a failed attempt in one place for the query and insert retry loops The retry decision after a failed attempt was written three times, once per retried operation (POJO insert, stream insert, query), so the cancellation fix of this PR had to touch all three copies - which SonarCloud reported as duplication on new code. The decision is now taken by one private helper: it rethrows the wrapped failure when no further request may be issued (a non-retryable failure, or an operation cancelled on the client side) and otherwise returns the endpoint of the next attempt. Behaviour is unchanged: the same exception instance is rethrown, and on the last attempt the node is still rotated without changing the endpoint. To keep the helper's parameter list short, an operation now carries the query id and the settings its attempts run with, and is therefore always created - it is only put into the cancellation registry when it has a query id to be addressed by. --- .../com/clickhouse/client/api/Client.java | 96 ++++++++++--------- 1 file changed, 53 insertions(+), 43 deletions(-) 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 a997d0b00..5f13bbf62 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 @@ -1464,7 +1464,7 @@ public CompletableFuture insert(String tableName, List data, requestSettings.setQueryId(queryIdGenerator.get()); } final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId); + final OngoingOperation ongoingOperation = registerOperation(queryId, requestSettings.getAllSettings()); Supplier supplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1508,15 +1508,7 @@ public CompletableFuture insert(String tableName, List data, } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(ongoingOperation)) { - if (i < maxAttempts) { - selectedEndpoint = logRetryAndSelectNextNode("Insert", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); - } else { - nodeSelector.getNextAliveNode(selectedEndpoint); - } - } else { - throw lastException; - } + selectedEndpoint = endpointForNextAttempt("Insert", ongoingOperation, e, lastException, i, maxAttempts, selectedEndpoint); } } } finally { @@ -1694,7 +1686,7 @@ public CompletableFuture insert(String tableName, final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings()); final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1); final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId); + final OngoingOperation ongoingOperation = registerOperation(queryId, requestSettings.getAllSettings()); Supplier responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1722,15 +1714,7 @@ public CompletableFuture insert(String tableName, } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(ongoingOperation)) { - if (i < maxAttempts) { - selectedEndpoint = logRetryAndSelectNextNode("Insert (stream)", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); - } else { - nodeSelector.getNextAliveNode(selectedEndpoint); - } - } else { - throw lastException; - } + selectedEndpoint = endpointForNextAttempt("Insert (stream)", ongoingOperation, e, lastException, i, maxAttempts, selectedEndpoint); } if (i < maxAttempts) { @@ -1837,7 +1821,7 @@ public CompletableFuture query(String sqlQuery, Map responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node @@ -1867,15 +1851,7 @@ public CompletableFuture query(String sqlQuery, Map settings) { + OngoingOperation operation = new OngoingOperation(queryId, settings); + if (queryId != null) { + // Without a query id the operation cannot be addressed by cancelTransportRequest(), so it is + // not registered - it still carries the state its own attempts need. + ongoingRequests.put(queryId, operation); } - OngoingOperation operation = new OngoingOperation(); - ongoingRequests.put(queryId, operation); return operation; } private void unregisterOperation(String queryId, OngoingOperation operation) { - if (queryId != null && operation != null) { + if (queryId != null) { // Removes only the entry of this operation: another one may have registered itself // under the same query id in the meantime. ongoingRequests.remove(queryId, operation); @@ -1921,13 +1917,11 @@ private void unregisterOperation(String queryId, OngoingOperation operation) { } private void registerTransportReq(OngoingOperation operation, TransportRequest tr) { - if (operation != null) { - operation.attach(tr); - } + operation.attach(tr); } private boolean requestIsNotCancelled(OngoingOperation operation) { - return operation == null || !operation.isCancelled(); + return !operation.isCancelled(); } /** @@ -1940,15 +1934,31 @@ private static RuntimeException cancelledException(RuntimeException lastExceptio } /** - * Cancellation state of a single operation. An operation issues one transport request per attempt, - * so cancellation is kept here rather than on the request: once an operation is cancelled it stays - * cancelled for every following attempt, and a request attached later is cancelled right away. + * State of a single operation: what identifies it, the settings its attempts run with, and whether it + * was cancelled. An operation issues one transport request per attempt, so cancellation is kept here + * rather than on the request: once an operation is cancelled it stays cancelled for every following + * attempt, and a request attached later is cancelled right away. */ private static final class OngoingOperation { + private final String queryId; + private final Map settings; private volatile boolean cancelled; private TransportRequest current; + OngoingOperation(String queryId, Map settings) { + this.queryId = queryId; + this.settings = settings; + } + + String queryId() { + return queryId; + } + + Map settings() { + return settings; + } + synchronized void attach(TransportRequest tr) { current = tr; if (cancelled) { From 5a65ecd7e5701c273911f499768ef630e81cf66b Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Sat, 1 Aug 2026 06:06:01 +0000 Subject: [PATCH 4/6] review: keep TransportRequest in the registry and drop OngoingOperation Addresses review feedback on #2990: the registry stays a ConcurrentHashMap and the OngoingOperation holder is removed. Instead the request of an attempt is unregistered in the operation scoped outer try/finally rather than per attempt, so it is still reachable in the window between two attempts, and the cancellation is checked at the top of every retry iteration. The cancellation that lands before the first request of an asynchronous operation was fixed by a second commit on this PR; that fix was built on the holder and is reverted here, to be raised separately. --- CHANGELOG.md | 9 +- .../com/clickhouse/client/api/Client.java | 213 ++++++------------ .../api/transport/TransportBaseTests.java | 106 +-------- docs/features.md | 2 +- 4 files changed, 83 insertions(+), 247 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b3598c8ee..977b20bc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,12 +46,9 @@ - **[client-v2]** Fixed `Client.cancelTransportRequest(queryId)` being silently dropped when it landed between two attempts of a retried operation (query, POJO insert and stream insert): the operation issued the next attempt anyway - and could complete successfully. Cancellation is now tracked per operation instead of per transport request, so a - cancelled operation stays cancelled for every following attempt and the retry loop stops instead of sending another - request. A cancellation that lands before the operation sent its first request is no longer dropped either - it was - possible to lose one with `useAsyncRequests(true)`, where the operation body runs on the shared executor: an operation - is now registered for cancellation when it is started rather than when its first request is created, so it stops - before any request reaches the server. (https://github.com/ClickHouse/clickhouse-java/issues/2989) + and could complete successfully. The request of an attempt now stays registered until the whole operation is over, + and the cancellation is checked before every attempt, so a cancelled operation stops instead of sending another + request. (https://github.com/ClickHouse/clickhouse-java/issues/2989) - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial read. (https://github.com/ClickHouse/clickhouse-java/issues/2985) 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 5f13bbf62..9e8cffad4 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 @@ -144,7 +144,7 @@ public class Client implements AutoCloseable { private final Map> typeHintMapping; - private final ConcurrentHashMap ongoingRequests = new ConcurrentHashMap<>(REQ_REGISTRY_SIZE); + private final ConcurrentHashMap ongoingRequests = new ConcurrentHashMap<>(REQ_REGISTRY_SIZE); // Server context private String dbUser; @@ -266,10 +266,6 @@ public void close() { LOG.debug("Skip closing operation executor because not owned by client"); } - // An operation is registered when it is started, so one that is still queued when the executor is - // shut down never runs and never removes its own registration. - ongoingRequests.clear(); - if (httpClientHelper != null) { httpClientHelper.close(); } @@ -1463,19 +1459,18 @@ public CompletableFuture insert(String tableName, List data, if (requestSettings.getQueryId() == null && queryIdGenerator != null) { requestSettings.setQueryId(queryIdGenerator.get()); } - final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId, requestSettings.getAllSettings()); Supplier supplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); + final String queryId = requestSettings.getQueryId(); RuntimeException lastException = null; try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled before or between - // attempts must not issue another request. - if (!requestIsNotCancelled(ongoingOperation)) { - throw cancelledException(lastException, queryId); + // A cancellation may have landed between two attempts: the request of the previous attempt is + // still registered, so it is seen here and no further request is issued. + if (!requestIsNotCancelled(queryId)) { + throw cancelledException(queryId, lastException); } // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), @@ -1498,7 +1493,7 @@ public CompletableFuture insert(String tableName, List data, out.close(); }); - registerTransportReq(ongoingOperation, transportRequest); + registerTransportReq(queryId, transportRequest); try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); @@ -1508,13 +1503,21 @@ public CompletableFuture insert(String tableName, List data, } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - selectedEndpoint = endpointForNextAttempt("Insert", ongoingOperation, e, lastException, i, maxAttempts, selectedEndpoint); + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(queryId)) { + if (i < maxAttempts) { + selectedEndpoint = logRetryAndSelectNextNode("Insert", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); + } else { + nodeSelector.getNextAliveNode(selectedEndpoint); + } + } else { + throw lastException; + } } } } finally { - // The operation is over: it will not issue another request, so its cancellation state - // is no longer needed. - unregisterOperation(queryId, ongoingOperation); + // The request of the last attempt stays registered until the operation is over, so a cancellation + // landing between two attempts is not lost. + unregisterTransportReq(queryId); } String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1522,7 +1525,7 @@ public CompletableFuture insert(String tableName, List data, throw (lastException == null ? new ClientException(errMsg) : lastException); }; - return runAsyncOperation(supplier, requestSettings.getAllSettings(), queryId, ongoingOperation); + return runAsyncOperation(supplier, requestSettings.getAllSettings()); } /** @@ -1685,20 +1688,20 @@ public CompletableFuture insert(String tableName, final int maxRetries = ClientConfigProperties.RETRY_ON_FAILURE.getOrDefault(requestSettings.getAllSettings()); final int maxAttempts = Math.max(maxRetries, endpoints.size() - 1); - final String queryId = requestSettings.getQueryId(); - final OngoingOperation ongoingOperation = registerOperation(queryId, requestSettings.getAllSettings()); Supplier responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); RuntimeException lastException = null; + final String queryId = requestSettings.getQueryId(); try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled before or between - // attempts (for instance from DataStreamWriter#onRetry()) must not issue another request. - if (!requestIsNotCancelled(ongoingOperation)) { - throw cancelledException(lastException, queryId); + // A cancellation may have landed between two attempts (for instance from DataStreamWriter#onRetry()): + // the request of the previous attempt is still registered, so it is seen here and no further + // request is issued. + if (!requestIsNotCancelled(queryId)) { + throw cancelledException(queryId, lastException); } // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), @@ -1706,7 +1709,7 @@ public CompletableFuture insert(String tableName, writer.onOutput(out); out.close(); }); - registerTransportReq(ongoingOperation, transportRequest); + registerTransportReq(queryId, transportRequest); try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { OperationMetrics metrics = completeOperation(transportResponse, finalClientStats, requestSettings.getQueryId()); @@ -1714,7 +1717,15 @@ public CompletableFuture insert(String tableName, } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); - selectedEndpoint = endpointForNextAttempt("Insert (stream)", ongoingOperation, e, lastException, i, maxAttempts, selectedEndpoint); + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(requestSettings.getQueryId())) { + if (i < maxAttempts) { + selectedEndpoint = logRetryAndSelectNextNode("Insert (stream)", i, maxAttempts, requestSettings.getQueryId(), selectedEndpoint, e); + } else { + nodeSelector.getNextAliveNode(selectedEndpoint); + } + } else { + throw lastException; + } } if (i < maxAttempts) { @@ -1726,7 +1737,9 @@ public CompletableFuture insert(String tableName, } } } finally { - unregisterOperation(queryId, ongoingOperation); + // The request of the last attempt stays registered until the operation is over, so a cancellation + // landing between two attempts is not lost. + unregisterTransportReq(queryId); } String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId()); @@ -1734,7 +1747,7 @@ public CompletableFuture insert(String tableName, throw (lastException == null ? new ClientException(errMsg) : lastException); }; - return runAsyncOperation(responseSupplier, requestSettings.getAllSettings(), queryId, ongoingOperation); + return runAsyncOperation(responseSupplier, requestSettings.getAllSettings()); } /** @@ -1820,22 +1833,21 @@ public CompletableFuture query(String sqlQuery, Map responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = nodeSelector.getEndpoint(); RuntimeException lastException = null; + final String queryId = requestSettings.getQueryId(); try { for (int i = 0; i <= maxAttempts; i++) { - // Cancellation belongs to the operation, so one that was cancelled before or between - // attempts must not issue another request. - if (!requestIsNotCancelled(ongoingOperation)) { - throw cancelledException(lastException, queryId); + // A cancellation may have landed between two attempts: the request of the previous attempt is + // still registered, so it is seen here and no further request is issued. + if (!requestIsNotCancelled(queryId)) { + throw cancelledException(queryId, lastException); } TransportRequest request = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), sqlQuery); - registerTransportReq(ongoingOperation, request); + registerTransportReq(queryId, request); TransportResponse transportResp = null; try { transportResp = httpClientHelper.executeRequest(request); @@ -1851,19 +1863,27 @@ public CompletableFuture query(String sqlQuery, Map settings) { - OngoingOperation operation = new OngoingOperation(queryId, settings); + private void unregisterTransportReq(String queryId) { if (queryId != null) { - // Without a query id the operation cannot be addressed by cancelTransportRequest(), so it is - // not registered - it still carries the state its own attempts need. - ongoingRequests.put(queryId, operation); + ongoingRequests.remove(queryId); } - return operation; } - private void unregisterOperation(String queryId, OngoingOperation operation) { + private boolean requestIsNotCancelled(String queryId) { if (queryId != null) { - // Removes only the entry of this operation: another one may have registered itself - // under the same query id in the meantime. - ongoingRequests.remove(queryId, operation); + TransportRequest tr = ongoingRequests.get(queryId); + return tr == null || !tr.isCancelled(); } - } - - private void registerTransportReq(OngoingOperation operation, TransportRequest tr) { - operation.attach(tr); - } - - private boolean requestIsNotCancelled(OngoingOperation operation) { - return !operation.isCancelled(); + return true; } /** - * Failure of an operation that was cancelled between two attempts. The same exception type and message + * Failure of an operation that was cancelled between two of its attempts. The same exception type and message * are used when a cancellation aborts an in-flight request, so a caller sees one outcome for a cancelled * operation regardless of when the cancellation landed. The failure of the last attempt is kept as cause. */ - private static RuntimeException cancelledException(RuntimeException lastException, String queryId) { + private static RuntimeException cancelledException(String queryId, RuntimeException lastException) { return new TransportException("Request was cancelled on client side", lastException, queryId); } - /** - * State of a single operation: what identifies it, the settings its attempts run with, and whether it - * was cancelled. An operation issues one transport request per attempt, so cancellation is kept here - * rather than on the request: once an operation is cancelled it stays cancelled for every following - * attempt, and a request attached later is cancelled right away. - */ - private static final class OngoingOperation { - - private final String queryId; - private final Map settings; - private volatile boolean cancelled; - private TransportRequest current; - - OngoingOperation(String queryId, Map settings) { - this.queryId = queryId; - this.settings = settings; - } - - String queryId() { - return queryId; - } - - Map settings() { - return settings; - } - - synchronized void attach(TransportRequest tr) { - current = tr; - if (cancelled) { - tr.cancel(); - } - } - - synchronized void cancel() { - cancelled = true; - if (current != null) { - current.cancel(); - } - } - - boolean isCancelled() { - return cancelled; - } - } - public CompletableFuture query(String sqlQuery, Map queryParams) { return query(sqlQuery, queryParams, null); } @@ -2348,22 +2298,6 @@ private String registerOperationMetrics() { return operationId; } - /** - * Runs an operation that registered itself in {@link #ongoingRequests} before being submitted, so - * {@link #cancelTransportRequest(String)} can reach it even before its first attempt starts. If the - * operation is never submitted its registration would never be removed by the operation itself, so it - * is dropped here. - */ - private CompletableFuture runAsyncOperation(Supplier resultSupplier, Map requestSettings, - String queryId, OngoingOperation operation) { - try { - return runAsyncOperation(resultSupplier, requestSettings); - } catch (RuntimeException | Error e) { - unregisterOperation(queryId, operation); - throw e; - } - } - private CompletableFuture runAsyncOperation(Supplier resultSupplier, Map requestSettings) { boolean isAsync = MapUtils.getFlag(requestSettings, configuration, ClientConfigProperties.ASYNC_OPERATIONS.getKey()); if (isAsync) { @@ -2509,17 +2443,16 @@ public void updateAccessToken(String accessToken) { * Tries to cancel ongoing request. This method cancels IO operations but doesn't * kill query on server side. Original queryId should be used to cancel the request. * This operation cancels only operations on client side and only that still waiting - * for response. Cancellation applies to the whole operation: it is effective as soon as the - * operation was started (so also for an asynchronous operation that has not sent its first - * request yet) and a retrying operation stops instead of issuing another attempt. + * for response. A cancellation that lands between two attempts of a retried operation is effective too: + * the operation stops instead of issuing another request. * * @param queryId - original query id that was passed in operation settings. */ public void cancelTransportRequest(String queryId) { Objects.requireNonNull(queryId, "queryId should be not null"); - OngoingOperation operation = ongoingRequests.get(queryId); - if (operation != null) { - operation.cancel(); + TransportRequest req = ongoingRequests.get(queryId); + if (req != null) { + req.cancel(); } } diff --git a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java index 5aac71df8..2bc8112f3 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -37,10 +37,6 @@ import java.util.Collections; import java.util.List; import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -591,9 +587,13 @@ public void onRetry() { int attempts = mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(); if (cancelOwnQueryId) { Assert.assertNotNull(opError, "[" + name + "] a cancelled operation must fail"); - Assert.assertTrue(opError instanceof TransportException - || opError.getCause() instanceof TransportException, + Throwable cancellation = opError instanceof TransportException ? opError : opError.getCause(); + Assert.assertTrue(cancellation instanceof TransportException, "[" + name + "] a cancelled operation must fail as cancelled, was: " + opError); + Assert.assertEquals(cancellation.getMessage(), "Request was cancelled on client side", + "[" + name + "] a cancelled operation must report the cancellation"); + Assert.assertNotNull(cancellation.getCause(), + "[" + name + "] the failure of the last attempt must be kept as cause"); Assert.assertEquals(attempts, 1, "[" + name + "] a cancelled operation must not issue another attempt"); @@ -621,100 +621,6 @@ public static Object[][] cancelBetweenAttemptsProvider() { }; } - /** - * An asynchronous operation runs its body on the shared executor, so a cancel issued right after the - * call returned can land before the operation has sent its first request. An operation is registered for - * cancellation when it is started, not when its first request is created, so such a cancel must stop it - * before any request reaches the server instead of being silently dropped. A cancel for an unrelated - * query id must leave the queued operation alone. - */ - @Test(groups = {"integration"}, dataProvider = "cancelBeforeFirstAttemptProvider") - public void testCancelBeforeFirstAttempt(String name, boolean insert, boolean cancelOwnQueryId) throws Exception { - if (isCloud()) { - return; // mocked server - } - - WireMockServer mockServer = startMockServer(); - mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) - .willReturn(WireMock.aResponse() - .withStatus(HttpStatus.SC_OK) - .withHeader("X-ClickHouse-Summary", - "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); - - ExecutorService operationExecutor = Executors.newSingleThreadExecutor(); - CountDownLatch executorBusy = new CountDownLatch(1); - CountDownLatch releaseExecutor = new CountDownLatch(1); - String queryId = "cancel-before-first-attempt-" + UUID.randomUUID(); - String cancelledQueryId = cancelOwnQueryId ? queryId : queryId + "-other"; - - try (Client client = new Client.Builder() - .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) - .setUsername("default") - .setPassword(ClickHouseServerForTest.getPassword()) - .compressClientRequest(false) - .compressServerResponse(false) - .useAsyncRequests(true) - .setSharedOperationExecutor(operationExecutor) - .build()) { - - // Occupies the only worker thread so the operation stays queued: it has not created its first - // request yet when the cancel is issued. - operationExecutor.submit(() -> { - executorBusy.countDown(); - return releaseExecutor.await(30, TimeUnit.SECONDS); - }); - Assert.assertTrue(executorBusy.await(30, TimeUnit.SECONDS), - "[" + name + "] operation executor did not start"); - - CompletableFuture operation = insert - ? client.insert("table01", out -> out.write("1\n".getBytes(StandardCharsets.US_ASCII)), - ClickHouseFormat.TSV, new InsertSettings().setQueryId(queryId)) - : client.query("SELECT 1", new QuerySettings().setQueryId(queryId)); - client.cancelTransportRequest(cancelledQueryId); - releaseExecutor.countDown(); - - Throwable opError = null; - try { - Object response = operation.get(30, TimeUnit.SECONDS); - Assert.assertNotNull(response); - ((AutoCloseable) response).close(); - } catch (Exception e) { - opError = e; - } - - int requests = mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(); - if (cancelOwnQueryId) { - Assert.assertNotNull(opError, "[" + name + "] a cancelled operation must fail"); - Throwable cancellation = opError instanceof TransportException ? opError : opError.getCause(); - Assert.assertTrue(cancellation instanceof TransportException, - "[" + name + "] a cancelled operation must fail as cancelled, was: " + opError); - Assert.assertTrue(cancellation.getMessage().contains("cancelled on client side"), - "[" + name + "] a cancelled operation must fail as cancelled, was: " + cancellation.getMessage()); - Assert.assertEquals(requests, 0, - "[" + name + "] a cancelled operation must not send a request"); - } else { - Assert.assertNull(opError, - "[" + name + "] an operation that was not cancelled must complete: " + opError); - Assert.assertEquals(requests, 1, - "[" + name + "] an operation that was not cancelled must send its request"); - } - } finally { - releaseExecutor.countDown(); - operationExecutor.shutdownNow(); - mockServer.stop(); - } - } - - @DataProvider(name = "cancelBeforeFirstAttemptProvider") - public static Object[][] cancelBeforeFirstAttemptProvider() { - return new Object[][]{ - {"query-cancelled", false, true}, - {"query-other-query-id", false, false}, - {"insert-cancelled", true, true}, - {"insert-other-query-id", true, false} - }; - } - /** * Reissues {@link Client#cancelTransportRequest(String)} until the worker stops or the timeout elapses. * Retrying makes the test robust against thread-scheduling races where a single cancel could land before diff --git a/docs/features.md b/docs/features.md index 7cedab081..fc5b3077b 100644 --- a/docs/features.md +++ b/docs/features.md @@ -33,7 +33,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t - Server information loading: Can refresh server version, current user, and server time zone information. - Compression support: Supports response compression, ClickHouse LZ4 request/response compression, HTTP content compression, and caller-supplied precompressed insert bodies. - Retry behavior: Can retry failed operations for configured failure causes and retry limits. -- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. Cancellation applies to the whole operation rather than to a single attempt: a cancelled operation that is being retried stops instead of issuing another request, including when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`) or before the first request was sent (an asynchronous operation runs its body on the shared executor, so a cancel issued right after the call returned can arrive first). +- Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. A cancelled operation that is being retried stops instead of issuing another request, also when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`). - Metrics and observability: Exposes client/server operation metrics and optionally integrates connection-pool gauges with Micrometer. - Configuration surface: Supports arbitrary client options, cookies, custom headers, server-setting prefixes, client naming, query id suppliers, and buffer sizing. - SQL helpers: Includes SQL quoting and temporal formatting helpers used by callers building SQL text safely. From e52f6554c63801f045ceb62fedbb1ad2ebe1f079 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Sat, 1 Aug 2026 06:40:14 +0000 Subject: [PATCH 5/6] refactor(client-v2): check the cancellation of a retried operation in one place The identical guard at the top of the POJO insert, stream insert and query retry loops is replaced by a single failIfCancelled(queryId, lastException) helper, in the same spirit as logRetryAndSelectNextNode(...). Behaviour is unchanged: the operation still fails with TransportException("Request was cancelled on client side") keeping the failure of the last attempt as cause. --- .../com/clickhouse/client/api/Client.java | 35 ++++++++----------- 1 file changed, 14 insertions(+), 21 deletions(-) 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 9e8cffad4..be9fd16ad 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 @@ -1467,11 +1467,7 @@ public CompletableFuture insert(String tableName, List data, RuntimeException lastException = null; try { for (int i = 0; i <= maxAttempts; i++) { - // A cancellation may have landed between two attempts: the request of the previous attempt is - // still registered, so it is seen here and no further request is issued. - if (!requestIsNotCancelled(queryId)) { - throw cancelledException(queryId, lastException); - } + failIfCancelled(queryId, lastException); // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), out -> { @@ -1697,12 +1693,7 @@ public CompletableFuture insert(String tableName, final String queryId = requestSettings.getQueryId(); try { for (int i = 0; i <= maxAttempts; i++) { - // A cancellation may have landed between two attempts (for instance from DataStreamWriter#onRetry()): - // the request of the previous attempt is still registered, so it is seen here and no further - // request is issued. - if (!requestIsNotCancelled(queryId)) { - throw cancelledException(queryId, lastException); - } + failIfCancelled(queryId, lastException); // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), out -> { @@ -1841,11 +1832,7 @@ public CompletableFuture query(String sqlQuery, Map query(String sqlQuery, Map queryParams) { From 43cdc47a14986c0873cc82b5d1e0d7045d8565dc Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Sat, 1 Aug 2026 07:09:35 +0000 Subject: [PATCH 6/6] review: do not check the cancellation before the first attempt The registry is keyed by query id, so before the first attempt of an operation a registered request can only belong to another operation that is still running: its cancellation must not stop the operation that is starting. --- .../com/clickhouse/client/api/Client.java | 13 ++-- .../api/transport/TransportBaseTests.java | 70 +++++++++++++++++++ 2 files changed, 78 insertions(+), 5 deletions(-) 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 be9fd16ad..2b022e393 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 @@ -1467,7 +1467,7 @@ public CompletableFuture insert(String tableName, List data, RuntimeException lastException = null; try { for (int i = 0; i <= maxAttempts; i++) { - failIfCancelled(queryId, lastException); + failIfCancelled(queryId, i, lastException); // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), out -> { @@ -1693,7 +1693,7 @@ public CompletableFuture insert(String tableName, final String queryId = requestSettings.getQueryId(); try { for (int i = 0; i <= maxAttempts; i++) { - failIfCancelled(queryId, lastException); + failIfCancelled(queryId, i, lastException); // Execute request TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), out -> { @@ -1832,7 +1832,7 @@ public CompletableFuture query(String sqlQuery, Map 0 && !requestIsNotCancelled(queryId)) { throw new TransportException("Request was cancelled on client side", lastException, queryId); } } diff --git a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java index 2bc8112f3..6d34d7e06 100644 --- a/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -37,6 +37,7 @@ import java.util.Collections; import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -621,6 +622,75 @@ public static Object[][] cancelBetweenAttemptsProvider() { }; } + /** + * The cancellation of a retried operation is looked up by query id, so it must not be evaluated before the + * first attempt: nothing of the starting operation is registered yet and a request found under the same query + * id belongs to another operation that is still running. Cancelling that other operation must not make the + * new one fail before it ever reaches the server. + */ + @Test(groups = {"integration"}) + public void testFirstAttemptNotStoppedByAnotherCancelledOperation() throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + + String queryId = "shared-query-id-" + UUID.randomUUID(); + CountDownLatch firstOperationStarted = new CountDownLatch(1); + CountDownLatch secondOperationDone = new CountDownLatch(1); + + try (Client client = mockServerClient(mockServer, 3)) { + // Holds the first operation in flight, and so registered, until the second one is over. + DataStreamWriter blockedWriter = out -> { + out.write("1\n".getBytes(StandardCharsets.US_ASCII)); + firstOperationStarted.countDown(); + try { + secondOperationDone.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + }; + DataStreamWriter plainWriter = out -> out.write("2\n".getBytes(StandardCharsets.US_ASCII)); + + Thread blockedOperation = new Thread(() -> { + try (InsertResponse ignored = client.insert("table01", blockedWriter, ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(60, TimeUnit.SECONDS)) { + // the outcome of the cancelled operation is not what this test is about + } catch (Exception expected) { + // cancelled or failed - either way the second operation is the subject here + } + }); + blockedOperation.setDaemon(true); + blockedOperation.start(); + Assert.assertTrue(firstOperationStarted.await(30, TimeUnit.SECONDS), + "the first operation should have reached the transport"); + client.cancelTransportRequest(queryId); + + Throwable error = null; + try (InsertResponse response = client.insert("table01", plainWriter, ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(30, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } catch (Exception e) { + error = e.getCause() == null ? e : e.getCause(); + } finally { + secondOperationDone.countDown(); + blockedOperation.join(30_000); + } + + Assert.assertNull(error, "an operation must not be stopped before its first attempt by the " + + "cancellation of another operation using the same query id, but failed with: " + error); + } finally { + mockServer.stop(); + } + } + /** * Reissues {@link Client#cancelTransportRequest(String)} until the worker stops or the timeout elapses. * Retrying makes the test robust against thread-scheduling races where a single cancel could land before