Skip to content

Fix client-v2: cancelTransportRequest lost between two retry attempts #2990

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
polyglotAI-bot wants to merge 6 commits into
base: main
Choose a base branch
from
+240 −41
Open

Fix client-v2: cancelTransportRequest lost between two retry attempts #2990

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
63633b5
fix(client-v2): keep cancellation per operation so it is not lost bet…
polyglotAI-bot Jul 31, 2026
f4ef787
fix(client-v2): register an operation for cancellation when it starts…
polyglotAI-bot Jul 31, 2026
dd3aab1
refactor(client-v2): handle a failed attempt in one place for the que…
polyglotAI-bot Jul 31, 2026
5a65ecd
review: keep TransportRequest in the registry and drop OngoingOperation
polyglotAI-bot Aug 1, 2026
e52f655
refactor(client-v2): check the cancellation of a retried operation in…
polyglotAI-bot Aug 1, 2026
43cdc47
review: do not check the cancellation before the first attempt
polyglotAI-bot Aug 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. 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)
Expand Down
104 changes: 64 additions & 40 deletions client-v2/src/main/java/com/clickhouse/client/api/Client.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1465,50 +1465,55 @@ public CompletableFuture<InsertResponse> 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);
try {
for (int i = 0; i <= maxAttempts; i++) {
failIfCancelled(queryId, i, lastException);
// 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(queryId, 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(queryId)) {
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 request of the last attempt stays registered until the operation is over, so a cancellation
// landing between two attempts is not lost.
unregisterTransportReq(queryId);
Comment on lines +1513 to +1516

@polyglotAI-bot polyglotAI-bot Aug 2, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks — the mismatch you spotted is real, and it is in the PR description, not the code: the description still described an earlier revision of this PR (an OngoingOperation holder with a sticky flag, removed with remove(key, value)), which was dropped in favour of the much smaller guard after @chernser asked to simplify. I have rewritten the description to match what is actually shipped, so it no longer promises identity-based removal.

On the substance: removing by query id only is pre-existing on main — unregisterTransportReq there is also ongoingRequests.remove(queryId), and registerTransportReq is a plain put, so two concurrent operations sharing a query id already alias each other today (the second put orphans the first entry, and either remove clears whichever entry is current). This PR does not introduce that; it only lengthens the window on the two insert paths, from one attempt to the whole operation, which is exactly what makes a between-attempts cancel visible at all.

I deliberately did not close it here:

  • A correct close is not just remove(queryId, request) in the finally — it also needs the registration side to be identity-checked and atomic against cancelTransportRequest (each loop carrying its own previous request, compute/computeIfPresent on both sides), i.e. a registry redesign across all three call sites.
  • None of its branches are reachable by a deterministic test (they require two operations racing on the same query id at microsecond granularity), so it would add uncovered new lines — the new-code coverage gate on this PR already went red once for that reason.
  • It runs against the simplification direction asked for in review.

It is now listed explicitly as a known residual in the PR description (item 3) rather than left implicit. Happy to implement the identity-checked registry — here or as a separate PR — if @chernser / @mshustov want it; leaving this thread open for that call.

}

String errMsg = requestExMsg("Insert", maxAttempts + 1, durationSince(startTime).toMillis(), requestSettings.getQueryId());
Expand Down Expand Up @@ -1688,6 +1693,7 @@ public CompletableFuture<InsertResponse> insert(String tableName,
final String queryId = requestSettings.getQueryId();
try {
for (int i = 0; i <= maxAttempts; i++) {
failIfCancelled(queryId, i, lastException);
// Execute request
TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(),
out -> {
Expand All @@ -1711,10 +1717,6 @@ public CompletableFuture<InsertResponse> 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) {
Expand All @@ -1726,6 +1728,8 @@ public CompletableFuture<InsertResponse> insert(String tableName,
}
}
} finally {
// 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);
}

Expand Down Expand Up @@ -1828,6 +1832,7 @@ public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Objec
final String queryId = requestSettings.getQueryId();
try {
for (int i = 0; i <= maxAttempts; i++) {
failIfCancelled(queryId, i, lastException);
TransportRequest request = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), sqlQuery);
registerTransportReq(queryId, request);
TransportResponse transportResp = null;
Expand Down Expand Up @@ -1901,6 +1906,24 @@ private boolean requestIsNotCancelled(String queryId) {
return true;
}

/**

@cursor cursor Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cancel lost before new attempt

Medium Severity

Cancellation for a retried operation is stored only on the current TransportRequest in ongoingRequests. After failIfCancelled passes at the start of a retry iteration, a concurrent cancelTransportRequest can mark that entry cancelled, but the next registerTransportReq replaces it with a fresh request, so the retry loop may still send another HTTP attempt despite the documented operation-wide cancel.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 43cdc47. Configure here.

@polyglotAI-bot polyglotAI-bot Aug 2, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks — the window is real, but I don't think it should be closed the way it first looks, so let me lay out what I checked.

Confirmed: between failIfCancelled(queryId, i, lastException) at the top of an iteration and the registerTransportReq(queryId, ...) a few lines later there is only httpClientHelper.createRequest(...). A cancelTransportRequest(queryId) landing inside that window marks the previous attempt's request, and the subsequent put replaces it, so the new attempt goes out. That is exactly the residual I flagged in the previous round ("query-id-keyed registry, put overwrites, no sticky flag").

But it is a pure thread race, not a reachable code path. Every deterministic between-attempts hook this client exposes — notably DataStreamWriter#onRetry() — runs at the end of the previous iteration, i.e. strictly before failIfCancelled, so it is caught by the guard (that is what TransportBaseTests#testCancelBetweenRetryAttempts pins). createRequest invokes no caller code: on the stream/POJO insert paths the output-stream lambda is only run later, from executeRequest. So there is no way to reach this window other than a concurrent canceller hitting a few microseconds of request construction, and no way to write a non-flaky regression test for it.

The obvious fix is wrong. Carrying cancellation over on registration (compute(queryId, (k, prev) -> prev != null && prev.isCancelled() ? prev : tr)) reintroduces precisely the defect you raised in the last round ("First-attempt cancel guard misfires", r3694834108): an unrelated operation's cancelled request left under the same query id would abort a brand-new operation. Closing this window correctly needs the swap to be identity-checked against this operation's own previous request and made atomic against cancelTransportRequest (both sides going through ConcurrentHashMap.compute/computeIfPresent so the cancel and the swap can't interleave):

// cancel side
ongoingRequests.computeIfPresent(queryId, (k, req) -> { req.cancel(); return req; });
// register side, attempts > 0 only
ongoingRequests.compute(queryId, (k, cur) -> (cur == previousAttemptRequest && cur.isCancelled()) ? cur : tr);

That is a real design change to the registry (each loop must carry its previous request; three call sites gain an abort branch), and @chernser's direction on this PR was explicitly to simplify — keep TransportRequest in the registry and just move unregistration to the outer try/finally. It also adds branches that no deterministic test can cover, which is what tripped the new-code coverage gate earlier on this PR.

So I'm deliberately not changing code here and leaving this thread open for @chernser / @mshustov to call: the PR as it stands makes cancellation effective for every deterministically reachable between-attempts case, and this remaining microsecond race is orthogonal to the bug in #2989. Happy to implement the identity-checked atomic registry above — in this PR or a follow-up — if you'd like it.

* Stops an operation that was cancelled before its next attempt is issued. A cancellation may have landed
* between two attempts (on the stream insert path for instance from {@link DataStreamWriter#onRetry()}): the
* request of the previous attempt stays registered until the operation is over, so it is seen here and no
* further request is issued.
* Only attempts that follow a failed one are checked: before the first attempt this operation has nothing
* registered yet, so a request found under the same query id belongs to another operation that is still
* running and its cancellation must not stop this one.
* 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. Called from the retry loop of every operation.
*/
private void failIfCancelled(String queryId, int attempt, RuntimeException lastException) {
if (attempt > 0 && !requestIsNotCancelled(queryId)) {
throw new TransportException("Request was cancelled on client side", lastException, queryId);
}
Comment on lines +1921 to +1924

@polyglotAI-bot polyglotAI-bot Aug 2, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct, and the doc mismatch is on us: the PR description still described an earlier revision of this change (the OngoingOperation holder with a sticky cancelled flag), which was removed in favour of the small failIfCancelled guard after @chernser asked to simplify. The description has been rewritten to describe the shipped design; docs/features.md and the CHANGELOG entry were already worded in terms of what the code does ("the request of an attempt stays registered until the operation is over, and the cancellation is checked before every attempt", best-effort), so no over-claim there — but please re-read the description, it was the stale part.

On the remaining race: this is the same finding @cursor[bot] raised at the guard site (r3699162510), answered there in detail. Summary: the window between failIfCancelled and registerTransportReq contains only httpClientHelper.createRequest(...) and no caller-visible hook (DataStreamWriter#onRetry() runs at the end of the previous iteration, strictly before the guard — that is what testCancelBetweenRetryAttempts pins), so it is not deterministically reachable and no non-flaky regression test can cover it.

The holder shape you propose is what this PR originally had, and it is the right end state — but note the naive carry-over version is not safe on its own: preserving a cancelled entry across registerTransportReq re-introduces the previous Bugbot finding on this PR ("First-attempt cancel guard misfires", r3694834108), where a foreign cancelled entry under the same query id aborts a brand-new operation. Doing it correctly means an identity-checked, atomic registry (compute/remove(key, value) on both the register and cancel sides), which also subsumes the aliasing point in your other comment.

That is a registry redesign with branches no deterministic test can exercise (the new-code coverage gate on this PR already went red once), so it is scoped out of this bug fix and now listed as known residuals 2 and 3 in the PR description. Leaving this thread unresolved so @chernser / @mshustov can decide whether they want it here or as a follow-up — I am happy to write it either way.

}

public CompletableFuture<QueryResponse> query(String sqlQuery, Map<String, Object> queryParams) {
return query(sqlQuery, queryParams, null);
}
Expand Down Expand Up @@ -2416,7 +2439,8 @@ 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. 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.
*/
Expand Down
Loading
Loading