From 7b2d5c1d90da11a5c0f4085dd89029a32449f1c3 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Wed, 24 Jun 2026 15:40:33 -0700 Subject: [PATCH 1/3] separated request creation and execution --- .../com/clickhouse/client/api/Client.java | 134 ++++++++-------- .../client/api/insert/InsertResponse.java | 10 +- .../client/api/internal/ClientUtils.java | 15 ++ .../api/internal/HttpAPIClientHelper.java | 146 ++++++++++++++---- .../client/api/query/QueryResponse.java | 44 +++--- .../transport/internal/TransportRequest.java | 36 +++++ .../transport/internal/TransportResponse.java | 60 +++++++ 7 files changed, 312 insertions(+), 133 deletions(-) create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportRequest.java create mode 100644 client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportResponse.java 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 da157cb9d..742d757a1 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 @@ -16,6 +16,7 @@ import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; import com.clickhouse.client.api.internal.ClientStatisticsHolder; +import com.clickhouse.client.api.internal.ClientUtils; import com.clickhouse.client.api.internal.CredentialsManager; import com.clickhouse.client.api.internal.HttpAPIClientHelper; import com.clickhouse.client.api.internal.MapUtils; @@ -36,6 +37,8 @@ import com.clickhouse.client.api.serde.POJOSerDe; import com.clickhouse.client.api.transport.Endpoint; import com.clickhouse.client.api.transport.HttpEndpoint; +import com.clickhouse.client.api.transport.internal.TransportRequest; +import com.clickhouse.client.api.transport.internal.TransportResponse; import com.clickhouse.client.config.ClickHouseClientOption; import com.clickhouse.data.ClickHouseColumn; import com.clickhouse.data.ClickHouseDataType; @@ -43,8 +46,6 @@ import com.google.common.collect.ImmutableList; import net.jpountz.lz4.LZ4Factory; import org.apache.hc.core5.concurrent.DefaultThreadFactory; -import org.apache.hc.core5.http.ClassicHttpResponse; -import org.apache.hc.core5.http.Header; import org.apache.hc.core5.http.HttpStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1319,43 +1320,37 @@ public CompletableFuture insert(String tableName, List data, RuntimeException lastException = null; for (int i = 0; i <= maxRetries; i++) { // Execute request - try (ClassicHttpResponse httpResponse = - httpClientHelper.executeRequest(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 | IOException e) { - throw new DataSerializationException(obj, serializer, e); - } - } + 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 | IOException e) { + throw new DataSerializationException(obj, serializer, e); } - out.close(); - })) { - - + } + } + out.close(); + }); + try (TransportResponse httpResponse = httpClientHelper.executeRequest(transportRequest)) { // Check response - if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime)); + if (httpResponse.getStatusCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { + LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getStatusCode(), durationSince(startTime)); selectedEndpoint = getNextAliveNode(); continue; } ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); - OperationMetrics metrics = new OperationMetrics(clientStats); - String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}"); - ProcessParser.parseSummary(summary, metrics); - String queryId = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID), requestSettings.getQueryId(), String::valueOf); - metrics.operationComplete(); - metrics.setQueryId(queryId); - return new InsertResponse(metrics, HttpAPIClientHelper.collectResponseHeaders(httpResponse)); + OperationMetrics metrics = completeOperation(httpResponse, clientStats, requestSettings.getQueryId()); + + return new InsertResponse(httpResponse, metrics); } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); @@ -1373,7 +1368,6 @@ public CompletableFuture insert(String tableName, List data, throw (lastException == null ? new ClientException(errMsg) : lastException); }; return runAsyncOperation(supplier, requestSettings.getAllSettings()); - } /** @@ -1509,7 +1503,7 @@ public CompletableFuture insert(String tableName, clientStats.start(ClientMetrics.OP_DURATION); final ClientStatisticsHolder finalClientStats = clientStats; - Supplier responseSupplier; + final int writeBufferSize = requestSettings.getInputStreamCopyBufferSize() <= 0 ? (int) configuration.get(ClientConfigProperties.CLIENT_NETWORK_BUFFER_SIZE.getKey()) : @@ -1533,7 +1527,8 @@ public CompletableFuture insert(String tableName, if (requestSettings.getQueryId() == null && queryIdGenerator != null) { requestSettings.setQueryId(queryIdGenerator.get()); } - responseSupplier = () -> { + + Supplier responseSupplier = () -> { long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = getNextAliveNode(); @@ -1541,28 +1536,23 @@ public CompletableFuture insert(String tableName, RuntimeException lastException = null; for (int i = 0; i <= retries; i++) { // Execute request - try (ClassicHttpResponse httpResponse = - httpClientHelper.executeRequest(selectedEndpoint, requestSettings.getAllSettings(), - out -> { - writer.onOutput(out); - out.close(); - })) { + TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), + out -> { + writer.onOutput(out); + out.close(); + }); + try (TransportResponse httpResponse = httpClientHelper.executeRequest(transportRequest)) { // Check response - if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime)); + if (httpResponse.getStatusCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { + LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getStatusCode(), durationSince(startTime)); selectedEndpoint = getNextAliveNode(); continue; } - OperationMetrics metrics = new OperationMetrics(finalClientStats); - String summary = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}"); - ProcessParser.parseSummary(summary, metrics); - String queryId = HttpAPIClientHelper.getHeaderVal(httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID), requestSettings.getQueryId(), String::valueOf); - metrics.operationComplete(); - metrics.setQueryId(queryId); - return new InsertResponse(metrics, HttpAPIClientHelper.collectResponseHeaders(httpResponse)); + OperationMetrics metrics = completeOperation(httpResponse, finalClientStats, requestSettings.getQueryId()); + return new InsertResponse(httpResponse, metrics); } catch (Exception e) { String msg = requestExMsg("Insert", (i + 1), durationSince(startTime).toMillis(), requestSettings.getQueryId()); lastException = httpClientHelper.wrapException(msg, e, requestSettings.getQueryId()); @@ -1668,38 +1658,28 @@ public CompletableFuture query(String sqlQuery, Map query(String sqlQuery, MapQueries data in one of descriptive format and creates a reader out of the response stream.

*

Format is selected internally so is ignored when passed in settings. If query contains format diff --git a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java index 2fc09f312..c83161725 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/insert/InsertResponse.java @@ -3,21 +3,17 @@ import com.clickhouse.client.api.http.ClickHouseHttpProto; import com.clickhouse.client.api.metrics.OperationMetrics; import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.transport.internal.TransportResponse; -import java.util.Collections; import java.util.Map; public class InsertResponse implements AutoCloseable { private OperationMetrics operationMetrics; private final Map responseHeaders; - public InsertResponse(OperationMetrics metrics) { - this(metrics, Collections.emptyMap()); - } - - public InsertResponse(OperationMetrics metrics, Map responseHeaders) { + public InsertResponse(TransportResponse transportResponse, OperationMetrics metrics) { this.operationMetrics = metrics; - this.responseHeaders = responseHeaders; + this.responseHeaders = transportResponse.getHeaders(); } @Override diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientUtils.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientUtils.java index 235d55cad..085c43181 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientUtils.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/ClientUtils.java @@ -1,5 +1,10 @@ package com.clickhouse.client.api.internal; +import org.slf4j.Logger; + +import java.io.Closeable; +import java.io.IOException; + /** * Class containing utility methods used across the client. */ @@ -14,4 +19,14 @@ public static boolean isNotBlank(String str) { public static boolean isBlank(String str) { return str == null || str.trim().isEmpty(); } + + public static void quiteClose(Closeable closeable, Logger log) { + if (closeable != null) { + try { + closeable.close(); + } catch (Exception e) { + log.warn("Failed to close object " + closeable, e); + } + } + } } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java index 37fe4ab8c..89c31ac09 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java @@ -13,6 +13,8 @@ import com.clickhouse.client.api.enums.ProxyType; import com.clickhouse.client.api.http.ClickHouseHttpProto; import com.clickhouse.client.api.transport.Endpoint; +import com.clickhouse.client.api.transport.internal.TransportRequest; +import com.clickhouse.client.api.transport.internal.TransportResponse; import com.clickhouse.client.config.ClickHouseDefaultSslContextProvider; import com.clickhouse.data.ClickHouseFormat; import net.jpountz.lz4.LZ4Factory; @@ -44,6 +46,7 @@ import org.apache.hc.core5.http.HttpRequest; import org.apache.hc.core5.http.HttpStatus; import org.apache.hc.core5.http.NoHttpResponseException; +import org.apache.hc.core5.http.ProtocolException; import org.apache.hc.core5.http.URIScheme; import org.apache.hc.core5.http.config.CharCodingConfig; import org.apache.hc.core5.http.config.Http1Config; @@ -513,9 +516,37 @@ private HttpPost createPostRequest(URI uri, Map requestConfig) { return req; } - public ClassicHttpResponse executeRequest(Endpoint server, Map requestConfig, - String body) throws Exception { + private static final class TransportRequestImpl implements TransportRequest { + private final HttpPost delegate; + private final Map config; + TransportRequestImpl(HttpPost delegate, Map config) { + this.delegate = delegate; + this.config = config; + } + + @Override + public boolean cancel() throws Exception { + if (delegate.isCancelled()) { + return true; + } + return delegate.cancel(); + } + + @Override + public Map getConfig() { + return config; + } + + @Override + @SuppressWarnings("unchecked") + public T getDelegate() { + return (T) delegate; + } + } + + public TransportRequest createRequest(Endpoint server, Map requestConfig, + String body) { boolean useMultipart = ClientConfigProperties.HTTP_SEND_PARAMS_IN_BODY.getOrDefault(requestConfig) && requestConfig.containsKey(HttpAPIClientHelper.KEY_STATEMENT_PARAMS); @@ -538,27 +569,93 @@ public ClassicHttpResponse executeRequest(Endpoint server, Map r req.setEntity(wrapRequestEntity(httpEntity, requestConfig)); } else { - final String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; - - HttpEntity httpEntity = new ByteArrayEntity(body.getBytes(StandardCharsets.UTF_8.name()), CONTENT_TYPE, contentEncoding); + final HttpEntity httpEntity; + try { + final String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; + httpEntity = new ByteArrayEntity(body.getBytes(StandardCharsets.UTF_8.name()), CONTENT_TYPE, contentEncoding); + } catch (UnsupportedEncodingException | ProtocolException e) { + throw new ClientException("failed to create request body entity", e); + } req.setEntity(wrapRequestEntity(httpEntity, requestConfig)); } - // execute - return doPostRequest(requestConfig, req); + return new TransportRequestImpl(req, requestConfig); } - public ClassicHttpResponse executeRequest(Endpoint server, Map requestConfig, - IOCallback writeCallback) throws Exception { + private static final class TransportResponseImpl implements TransportResponse { + + private final ClassicHttpResponse delegate; + + TransportResponseImpl(ClassicHttpResponse delegate) { + this.delegate = delegate; + } + + @Override + public int getStatusCode() { + return 0; + } + + @Override + public ClickHouseFormat getDataFormat() { + Header formatHeader = delegate.getFirstHeader(ClickHouseHttpProto.HEADER_FORMAT); + return formatHeader == null ? null : ClickHouseFormat.valueOf(formatHeader.getValue()); + } + + @Override + public String getSummaryJson() { + return HttpAPIClientHelper.getHeaderVal(delegate + .getFirstHeader(ClickHouseHttpProto.HEADER_SRV_SUMMARY), "{}"); + } + + @Override + public String getQueryId() { + return HttpAPIClientHelper.getHeaderVal(delegate + .getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID), null); + } + + @Override + public T getDelegate() { + return (T) delegate; + } + + @Override + public Map getHeaders() { + return HttpAPIClientHelper.collectResponseHeaders(delegate); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + + @Override + public InputStream createDataInputStream() { + try { + return delegate.getEntity().getContent(); + } catch (Exception e) { + throw new ClientException("Failed to construct input stream", e); + } + } + } + + public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception { + return new TransportResponseImpl(doPostRequest(transportRequest.getConfig(), transportRequest.getDelegate())); + } + + public TransportRequest createRequest(Endpoint server, Map requestConfig, IOCallback writeCallback) { final URI uri = createRequestURI(server, requestConfig, true); final HttpPost req = createPostRequest(uri, requestConfig); - String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; - req.setEntity(wrapRequestEntity( - new EntityTemplate(-1, CONTENT_TYPE, contentEncoding , writeCallback), - requestConfig)); + try { + String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null; + req.setEntity(wrapRequestEntity( + new EntityTemplate(-1, CONTENT_TYPE, contentEncoding, writeCallback), + requestConfig)); + } catch (ProtocolException e) { + throw new ClientException("failed to create request body entity", e); + } - return doPostRequest(requestConfig, req); + return new TransportRequestImpl(req, requestConfig); } private ClassicHttpResponse doPostRequest(Map requestConfig, HttpPost req) throws Exception { @@ -589,30 +686,20 @@ private ClassicHttpResponse doPostRequest(Map requestConfig, Htt return httpResponse; } catch (UnknownHostException e) { - closeQuietly(httpResponse); + ClientUtils.quiteClose(httpResponse, LOG); LOG.warn("Host '{}' unknown", req.getAuthority()); throw e; } catch (ConnectException | NoRouteToHostException e) { - closeQuietly(httpResponse); + ClientUtils.quiteClose(httpResponse, LOG); LOG.warn("Failed to connect to '{}': {}", req.getAuthority(), e.getMessage()); throw e; } catch (Exception e) { - closeQuietly(httpResponse); + ClientUtils.quiteClose(httpResponse, LOG); LOG.debug("Failed to execute request to '{}': {}", req.getAuthority(), e.getMessage(), e); throw e; } } - public static void closeQuietly(ClassicHttpResponse httpResponse) { - if (httpResponse != null) { - try { - httpResponse.close(); - } catch (IOException e) { - LOG.warn("Failed to close response"); - } - } - } - private static final ContentType CONTENT_TYPE = ContentType.create(ContentType.TEXT_PLAIN.getMimeType(), "UTF-8"); private void addHeaders(HttpPost req, Map requestConfig) { @@ -804,7 +891,10 @@ public static int getHeaderInt(Header header, int defaultValue) { ClickHouseHttpProto.HEADER_SRV_SUMMARY, ClickHouseHttpProto.HEADER_SRV_DISPLAY_NAME, ClickHouseHttpProto.HEADER_DATABASE, - ClickHouseHttpProto.HEADER_DB_USER + ClickHouseHttpProto.HEADER_DB_USER, + ClickHouseHttpProto.HEADER_TIMEZONE, + ClickHouseHttpProto.HEADER_FORMAT, + ClickHouseHttpProto.HEADER_PROGRESS )); /** diff --git a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java index 6e237a18d..32c1d4dab 100644 --- a/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java +++ b/client-v2/src/main/java/com/clickhouse/client/api/query/QueryResponse.java @@ -3,14 +3,14 @@ import com.clickhouse.client.api.ClientConfigProperties; import com.clickhouse.client.api.ClientException; import com.clickhouse.client.api.http.ClickHouseHttpProto; +import com.clickhouse.client.api.internal.ClientUtils; import com.clickhouse.client.api.metrics.OperationMetrics; import com.clickhouse.client.api.metrics.ServerMetrics; +import com.clickhouse.client.api.transport.internal.TransportResponse; import com.clickhouse.data.ClickHouseFormat; -import org.apache.hc.core5.http.ClassicHttpResponse; -import org.apache.hc.core5.http.Header; import java.io.InputStream; -import java.util.Collections; +import java.time.ZoneId; import java.util.Map; import java.util.TimeZone; @@ -31,51 +31,43 @@ public class QueryResponse implements AutoCloseable { private final ClickHouseFormat format; - private QuerySettings settings; + private final QuerySettings settings; - private OperationMetrics operationMetrics; + private final OperationMetrics operationMetrics; - private ClassicHttpResponse httpResponse; + private final TransportResponse transportResponse; private final Map responseHeaders; - public QueryResponse(ClassicHttpResponse response, ClickHouseFormat format, QuerySettings settings, - OperationMetrics operationMetrics) { - this(response, format, settings, operationMetrics, Collections.emptyMap()); - } - - public QueryResponse(ClassicHttpResponse response, ClickHouseFormat format, QuerySettings settings, - OperationMetrics operationMetrics, Map responseHeaders) { - this.httpResponse = response; + public QueryResponse(TransportResponse response, ClickHouseFormat format, QuerySettings settings, OperationMetrics operationMetrics) { + this.transportResponse = response; this.format = format; this.operationMetrics = operationMetrics; this.settings = settings; - this.responseHeaders = responseHeaders; + this.responseHeaders = response.getHeaders(); - Header tzHeader = response.getFirstHeader(ClickHouseHttpProto.HEADER_TIMEZONE); - if (tzHeader != null) { + String timeZoneHeader = responseHeaders.get(ClickHouseHttpProto.HEADER_TIMEZONE); + if (timeZoneHeader != null) { + TimeZone serverTz; try { - this.settings.setOption(ClientConfigProperties.SERVER_TIMEZONE.getKey(), - TimeZone.getTimeZone(tzHeader.getValue())); + serverTz = TimeZone.getTimeZone(timeZoneHeader); } catch (Exception e) { throw new ClientException("Failed to parse server timezone", e); } + this.settings.setOption(ClientConfigProperties.SERVER_TIMEZONE.getKey(), + serverTz); } } public InputStream getInputStream() { - try { - return httpResponse.getEntity().getContent(); - } catch (Exception e) { - throw new ClientException("Failed to construct input stream", e); - } + return transportResponse.createDataInputStream(); } @Override public void close() throws Exception { - if (httpResponse != null ) { + if (transportResponse != null ) { try { - httpResponse.close(); + transportResponse.close(); } catch (Exception e) { throw new ClientException("Failed to close response", e); } diff --git a/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportRequest.java b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportRequest.java new file mode 100644 index 000000000..6a1a66401 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportRequest.java @@ -0,0 +1,36 @@ +package com.clickhouse.client.api.transport.internal; + +import java.util.Map; + +public interface TransportRequest { + + + /** + * Gives access to transport delegate. Used strictly only by transport. + * @return internal transport request object + * @param - Type of delegate + */ + T getDelegate(); + + /** + * Returns reference to request configuration. Implementation should + * store only copy because configuration map is created for each request separately + * @return request configuration map + */ + Map getConfig(); + + /** + * Cancels request associated with the object. Implementation should + * treat this method like close() and release all resource. + * When request is canceled it cannot be reused. All reusable objects + * should be saved elsewhere. + * In many cases cancellation of IO operation is problematic and result + * of this method should not be used in core logic. + * Operation is idempotent and can be called on canceled request multiple + * times. + * + * @return result of operation. True if request was canceled for sure. False when result cannot be known. + * @throws Exception - when something extraordinary happens while canceling the request. + */ + boolean cancel() throws Exception; +} diff --git a/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportResponse.java b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportResponse.java new file mode 100644 index 000000000..32fe89465 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportResponse.java @@ -0,0 +1,60 @@ +package com.clickhouse.client.api.transport.internal; + +import com.clickhouse.data.ClickHouseFormat; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Map; + +public interface TransportResponse extends Closeable { + + /** + * Transport status code translated to one of values: + *

    + *
  • 503 - for service unavailable
    • + *
  • 500 - server error
    • + *
  • 400 - user error
    • + *
  • 404 - endpoint not found
    • + *
  • 403 - access not granted
    • + *
  • 401 - no authentication information
    • + *
  • 200 - ok
    • + *
+ * @return integer value of status code + */ + int getStatusCode(); + + + /** + * Data format returned by server or calculated other way + * @return data format + */ + ClickHouseFormat getDataFormat(); + + String getSummaryJson(); + + String getQueryId(); + + /** + * Gives access to transport delegate. Used strictly only by transport. + * @return internal transport response object + * @param - Type of delegate + */ + T getDelegate(); + + + /** + * Server headers. + * @return response headers + */ + Map getHeaders(); + + + /** + * Creates a new stream to read data. It is applicable only for + * blocking transports. In real life this should be called once. + * It is important to mention that each time new input stream is created. + * + * @return new data stream + */ + InputStream createDataInputStream(); +} From fd444b06449506a456ca5a1dfe993d1091846722 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Thu, 25 Jun 2026 22:00:56 -0700 Subject: [PATCH 2/3] Added more tests and removed dead code halding 503 in client --- .../com/clickhouse/client/api/Client.java | 39 +-- .../client/api/query/QueryResponse.java | 14 +- .../client/api/internal/ClientUtilsTest.java | 48 +++ .../api/transport/TransportBaseTests.java | 326 ++++++++++++++++++ .../clickhouse/client/insert/InsertTests.java | 31 ++ .../clickhouse/client/query/QueryTests.java | 49 +++ 6 files changed, 468 insertions(+), 39 deletions(-) create mode 100644 client-v2/src/test/java/com/clickhouse/client/api/internal/ClientUtilsTest.java create mode 100644 client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java 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 ba5a95881..6f0a56be7 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 @@ -47,7 +47,6 @@ import com.google.common.collect.ImmutableList; import net.jpountz.lz4.LZ4Factory; import org.apache.hc.core5.concurrent.DefaultThreadFactory; -import org.apache.hc.core5.http.HttpStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1398,18 +1397,11 @@ public CompletableFuture insert(String tableName, List data, } out.close(); }); - try (TransportResponse httpResponse = httpClientHelper.executeRequest(transportRequest)) { - // Check response - if (httpResponse.getStatusCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getStatusCode(), durationSince(startTime)); - selectedEndpoint = getNextAliveNode(); - continue; - } - + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { ClientStatisticsHolder clientStats = globalClientStats.remove(operationId); - OperationMetrics metrics = completeOperation(httpResponse, clientStats, requestSettings.getQueryId()); + OperationMetrics metrics = completeOperation(transportResponse, clientStats, requestSettings.getQueryId()); - return new InsertResponse(httpResponse, metrics); + 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()); @@ -1424,7 +1416,8 @@ public CompletableFuture insert(String tableName, List data, String errMsg = requestExMsg("Insert", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId()); LOG.warn(errMsg); - throw (lastException == null ? new ClientException(errMsg) : lastException); }; + throw (lastException == null ? new ClientException(errMsg) : lastException); + }; return runAsyncOperation(supplier, requestSettings.getAllSettings()); } @@ -1601,17 +1594,9 @@ public CompletableFuture insert(String tableName, out.close(); }); - try (TransportResponse httpResponse = httpClientHelper.executeRequest(transportRequest)) { - - // Check response - if (httpResponse.getStatusCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getStatusCode(), durationSince(startTime)); - selectedEndpoint = getNextAliveNode(); - continue; - } - - OperationMetrics metrics = completeOperation(httpResponse, finalClientStats, requestSettings.getQueryId()); - return new InsertResponse(httpResponse, metrics); + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { + OperationMetrics metrics = completeOperation(transportResponse, finalClientStats, 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()); @@ -1722,14 +1707,6 @@ public CompletableFuture query(String sqlQuery, Map responseHeaders; public QueryResponse(TransportResponse response, ClickHouseFormat format, QuerySettings settings, OperationMetrics operationMetrics) { + Objects.requireNonNull(response, "response is null"); this.transportResponse = response; this.format = format; this.operationMetrics = operationMetrics; @@ -65,12 +65,10 @@ public InputStream getInputStream() { @Override public void close() throws Exception { - if (transportResponse != null ) { - try { - transportResponse.close(); - } catch (Exception e) { - throw new ClientException("Failed to close response", e); - } + try { + transportResponse.close(); + } catch (Exception e) { + throw new ClientException("Failed to close response", e); } } diff --git a/client-v2/src/test/java/com/clickhouse/client/api/internal/ClientUtilsTest.java b/client-v2/src/test/java/com/clickhouse/client/api/internal/ClientUtilsTest.java new file mode 100644 index 000000000..9cf10b25a --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/internal/ClientUtilsTest.java @@ -0,0 +1,48 @@ +package com.clickhouse.client.api.internal; + +import org.mockito.Mockito; +import org.slf4j.Logger; +import org.testng.Assert; +import org.testng.annotations.Test; + +import java.io.Closeable; +import java.io.IOException; + +public class ClientUtilsTest { + + @Test(groups = {"unit"}) + public void testQuiteCloseSwallowsExceptionAndLogs() throws IOException { + Logger log = Mockito.mock(Logger.class); + IOException failure = new IOException("close failed"); + Closeable closeable = Mockito.mock(Closeable.class); + Mockito.doThrow(failure).when(closeable).close(); + + // Should not propagate the exception thrown by close() + ClientUtils.quiteClose(closeable, log); + + Mockito.verify(closeable).close(); + Mockito.verify(log).warn(Mockito.contains("Failed to close object"), Mockito.eq(failure)); + } + + @Test(groups = {"unit"}) + public void testQuiteCloseClosesSuccessfully() throws IOException { + Logger log = Mockito.mock(Logger.class); + Closeable closeable = Mockito.mock(Closeable.class); + + ClientUtils.quiteClose(closeable, log); + + Mockito.verify(closeable).close(); + Mockito.verifyNoInteractions(log); + } + + @Test(groups = {"unit"}) + public void testQuiteCloseWithNull() { + Logger log = Mockito.mock(Logger.class); + + // Should be a no-op and not throw on a null closeable + ClientUtils.quiteClose(null, log); + + Mockito.verifyNoInteractions(log); + Assert.assertTrue(true); + } +} 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 new file mode 100644 index 000000000..113304e6c --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -0,0 +1,326 @@ +package com.clickhouse.client.api.transport; + + +import com.clickhouse.client.BaseIntegrationTest; +import com.clickhouse.client.ClickHouseServerForTest; +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.ClientFaultCause; +import com.clickhouse.client.api.ServerException; +import com.clickhouse.client.api.enums.Protocol; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.query.QueryResponse; +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.client.WireMock; +import com.github.tomakehurst.wiremock.common.ConsoleNotifier; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseFormat; +import org.apache.hc.core5.http.HttpStatus; +import org.testcontainers.utility.ThrowingFunction; +import org.testng.Assert; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.io.ByteArrayInputStream; +import java.util.Collections; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED; + +@Test(groups = {"integration"}) +public class TransportBaseTests extends BaseIntegrationTest { + + private static final String RETRYABLE_BODY = + "Code: 202. DB::Exception: Too many simultaneous queries. (TOO_MANY_SIMULTANEOUS_QUERIES)"; + private static final int RETRYABLE_CODE = 202; + + private static final String NON_RETRYABLE_BODY = + "Code: 62. DB::Exception: Syntax error. (SYNTAX_ERROR)"; + private static final int NON_RETRYABLE_CODE = 62; + + private WireMockServer startMockServer() { + WireMockServer mockServer = new WireMockServer(WireMockConfiguration + .options().dynamicPort().notifier(new ConsoleNotifier(false))); + mockServer.start(); + return mockServer; + } + + private Client mockServerClient(WireMockServer mockServer, int maxRetries, ClientFaultCause... retryOn) { + Client.Builder builder = new Client.Builder() + .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressClientRequest(false) + .compressServerResponse(false) + .setMaxRetries(maxRetries); + if (retryOn.length > 0) { + builder.retryOnFailures(retryOn); + } + return builder.build(); + } + + /** + * A retryable server error on the first attempt should be retried and the following successful + * response should be returned. Covers the retry branch of the operation loop + * ({@code shouldRetry(...)} == true) and recovery on the next node. + */ + @Test(groups = {"integration"}, dataProvider = "operationProvider") + public void testRetriesAndSucceedsAfterRetryableServerError(String operation, + ThrowingFunction function) { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("Retry") + .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("Retry") + .whenScenarioStateIs("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + + try (Client client = mockServerClient(mockServer, 1)) { + function.apply(client); + } catch (Exception e) { + Assert.fail("[" + operation + "] should have recovered after a retry", e); + } finally { + mockServer.stop(); + } + + Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(), 2, + "[" + operation + "] expected one failed attempt followed by a successful retry"); + } + + /** + * When the server keeps returning a retryable error the operation should be retried + * {@code maxRetries} times and then fail by re-throwing the last captured exception. + * Covers the final throw of the operation loop ({@code throw lastException}). + */ + @Test(groups = {"integration"}, dataProvider = "operationProvider") + public void testThrowsAfterExhaustingRetries(String operation, + ThrowingFunction function) { + if (isCloud()) { + return; // mocked server + } + + int maxRetries = 2; + WireMockServer mockServer = startMockServer(); + + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(RETRYABLE_CODE)) + .withBody(RETRYABLE_BODY)).build()); + + try (Client client = mockServerClient(mockServer, maxRetries)) { + function.apply(client); + Assert.fail("[" + operation + "] expected exception after exhausting retries"); + } catch (ServerException e) { + Assert.assertEquals(e.getCode(), RETRYABLE_CODE); + } catch (Exception e) { + Assert.fail("[" + operation + "] unexpected exception type", e); + } finally { + mockServer.stop(); + } + + Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(), maxRetries + 1, + "[" + operation + "] expected initial attempt plus " + maxRetries + " retries"); + } + + /** + * A non-retryable server error should be re-thrown immediately without consuming any of the + * configured retries. Covers the {@code else { throw lastException; }} branch. + */ + @Test(groups = {"integration"}, dataProvider = "operationProvider") + public void testThrowsImmediatelyWhenNotRetryable(String operation, + ThrowingFunction function) { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_BAD_REQUEST) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(NON_RETRYABLE_CODE)) + .withBody(NON_RETRYABLE_BODY)).build()); + + try (Client client = mockServerClient(mockServer, 3)) { + function.apply(client); + Assert.fail("[" + operation + "] expected exception for non-retryable error"); + } catch (ServerException e) { + Assert.assertEquals(e.getCode(), NON_RETRYABLE_CODE); + } catch (Exception e) { + Assert.fail("[" + operation + "] unexpected exception type", e); + } finally { + mockServer.stop(); + } + + Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(), 1, + "[" + operation + "] a non-retryable error must not be retried"); + } + + /** + * When retries are disabled ({@link ClientFaultCause#None}) even an otherwise retryable error + * must be re-thrown on the first attempt. Covers the {@code else { throw lastException; }} + * branch when {@code shouldRetry(...)} returns {@code false} because of the configuration. + */ + @Test(groups = {"integration"}, dataProvider = "operationProvider") + public void testRetriesDisabledThrowsImmediately(String operation, + ThrowingFunction function) { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = startMockServer(); + + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", String.valueOf(RETRYABLE_CODE)) + .withBody(RETRYABLE_BODY)).build()); + + try (Client client = mockServerClient(mockServer, 3, ClientFaultCause.None)) { + function.apply(client); + Assert.fail("[" + operation + "] expected exception when retries are disabled"); + } catch (ServerException e) { + Assert.assertEquals(e.getCode(), RETRYABLE_CODE); + } catch (Exception e) { + Assert.fail("[" + operation + "] unexpected exception type", e); + } finally { + mockServer.stop(); + } + + Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(), 1, + "[" + operation + "] retries are disabled, no retry expected"); + } + + /** + * When no query id is supplied by the caller but a generator is configured, the operation must + * use the generated id. Covers the + * {@code if (requestSettings.getQueryId() == null && queryIdGenerator != null)} block that is + * present in every operation path. + */ + @Test(groups = {"integration"}, dataProvider = "operationProvider") + public void testGeneratedQueryIdIsUsedWhenNotSet(String operation, + ThrowingFunction function) { + 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 generatedId = "generated-" + UUID.randomUUID(); + AtomicInteger generatorCalls = new AtomicInteger(); + Supplier idGenerator = () -> { + generatorCalls.incrementAndGet(); + return generatedId; + }; + + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressClientRequest(false) + .compressServerResponse(false) + .setQueryIdGenerator(idGenerator) + .build()) { + function.apply(client); + } catch (Exception e) { + mockServer.stop(); + Assert.fail("[" + operation + "] operation should succeed", e); + return; + } + + try { + Assert.assertTrue(generatorCalls.get() >= 1, + "[" + operation + "] query id generator should have been invoked"); + mockServer.verify(WireMock.postRequestedFor(WireMock.anyUrl()) + .withQueryParam("query_id", WireMock.equalTo(generatedId))); + } finally { + mockServer.stop(); + } + } + + /** + * Provides one {@link ThrowingFunction} per retry-loop code path in {@link Client}, so every + * failure-handling test exercises all of them: + *
    + *
  • {@code query} - {@link Client#query(String)}
    • + *
  • {@code insert-stream} - {@link Client#insert(String, java.io.InputStream, ClickHouseFormat)}
    • + *
  • {@code insert-pojo} - {@link Client#insert(String, java.util.List)}
    • + *
+ */ + @DataProvider(name = "operationProvider") + public static Object[][] operationProvider() { + ThrowingFunction queryFunction = (client) -> { + try (QueryResponse response = client.query("SELECT timezone()").get(30, TimeUnit.SECONDS)) { + return null; + } + }; + + ThrowingFunction streamInsertFunction = (client) -> { + try (InsertResponse response = client.insert("table01", + new ByteArrayInputStream("1\t2\t3\n".getBytes()), ClickHouseFormat.TSV) + .get(30, TimeUnit.SECONDS)) { + return null; + } + }; + + ThrowingFunction pojoInsertFunction = (client) -> { + client.register(InsertablePojo.class, new TableSchema("table01", null, "default", + Collections.singletonList(ClickHouseColumn.of("id", "Int32")))); + try (InsertResponse response = client.insert("table01", + Collections.singletonList(new InsertablePojo(1))) + .get(30, TimeUnit.SECONDS)) { + return null; + } + }; + + return new Object[][]{ + {"query", queryFunction}, + {"insert-stream", streamInsertFunction}, + {"insert-pojo", pojoInsertFunction} + }; + } + + public static class InsertablePojo { + private int id; + + public InsertablePojo() { + } + + public InsertablePojo(int id) { + this.id = id; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + } +} diff --git a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java index 6ffdfea5e..cf9eb057b 100644 --- a/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/insert/InsertTests.java @@ -13,6 +13,7 @@ import com.clickhouse.client.api.enums.Protocol; import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.internal.ClientUtils; import com.clickhouse.client.api.internal.ServerSettings; import com.clickhouse.client.api.metadata.TableSchema; import com.clickhouse.client.api.metrics.ClientMetrics; @@ -140,6 +141,36 @@ public void insertSimplePOJOs() throws Exception { assertEquals(response.getQueryId(), uuid); } + @Test(groups = { "integration" }, enabled = true) + public void testInsertResponseStats() throws Exception { + String tableName = "insert_response_stats_table"; + String createSQL = SamplePOJO.generateTableCreateSQL(tableName); + String uuid = UUID.randomUUID().toString(); + + initTable(tableName, createSQL); + + client.register(SamplePOJO.class, client.getTableSchema(tableName)); + List simplePOJOs = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + simplePOJOs.add(new SamplePOJO()); + } + + settings.setQueryId(uuid); + try (InsertResponse response = client.insert(tableName, simplePOJOs, settings).get(EXECUTE_CMD_TIMEOUT, TimeUnit.SECONDS)) { + assertEquals(response.getWrittenRows(), simplePOJOs.size()); + assertTrue(response.getWrittenBytes() > 0, "written bytes should be positive"); + // server reads back the inserted rows to write them, so read counters are populated + assertTrue(response.getReadRows() >= 0, "read rows should be non-negative"); + assertTrue(response.getReadBytes() >= 0, "read bytes should be non-negative"); + assertTrue(response.getServerTime() >= 0, "server time should be non-negative"); + assertTrue(response.getResultRows() >= 0, "result rows should be non-negative"); + assertEquals(response.getQueryId(), uuid); + Assert.assertNotNull(response.getMetrics(), "metrics should be present"); + Assert.assertTrue(ClientUtils.isNotBlank(response.getServerDisplayName())); + Assert.assertNotNull(response.getResponseHeaders(), "response headers should be present"); + } + } + @Test(groups = { "integration" }, enabled = true) public void insertPOJOWithJSON() throws Exception { if (isCloud()) { diff --git a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java index d9db0d0ce..94a4c490b 100644 --- a/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java @@ -11,6 +11,7 @@ import com.clickhouse.client.api.ClientException; import com.clickhouse.client.api.ServerException; import com.clickhouse.client.api.command.CommandSettings; +import com.clickhouse.client.api.http.ClickHouseHttpProto; import com.clickhouse.client.api.data_formats.ClickHouseBinaryFormatReader; import com.clickhouse.client.api.data_formats.internal.BinaryStreamReader; import com.clickhouse.client.api.enums.Protocol; @@ -1466,6 +1467,54 @@ public void testQueryMetrics() throws Exception { } } + @Test(groups = {"integration"}) + public void testQueryResponseGettersAndHeaders() throws Exception { + String uuid = UUID.randomUUID().toString(); + QuerySettings settings = new QuerySettings() + .setFormat(ClickHouseFormat.TabSeparated) + .waitEndOfQuery(true) + .setQueryId(uuid); + + try (QueryResponse response = client.query("SELECT number FROM system.numbers LIMIT 10", settings).get()) { + // Format getter + Assert.assertEquals(response.getFormat(), ClickHouseFormat.TabSeparated); + + // Metrics getters + Assert.assertNotNull(response.getMetrics(), "metrics should be present"); + Assert.assertEquals(response.getReadRows(), 10); + Assert.assertTrue(response.getReadBytes() > 0, "read bytes should be positive"); + Assert.assertTrue(response.getWrittenRows() >= 0, "written rows should be non-negative"); + Assert.assertTrue(response.getWrittenBytes() >= 0, "written bytes should be non-negative"); + Assert.assertTrue(response.getServerTime() >= 0, "server time should be non-negative"); + Assert.assertTrue(response.getResultRows() > 0, "result rows should be positive"); + Assert.assertTrue(response.getTotalRowsToRead() >= 0, "total rows to read should be non-negative"); + Assert.assertEquals(response.getQueryId(), uuid); + + // Timezone and settings getters + Assert.assertNotNull(response.getTimeZone(), "server timezone should be resolved from response header"); + Assert.assertNotNull(response.getSettings(), "settings should be present"); + + // Server display name getter + Assert.assertNotNull(response.getServerDisplayName(), "server display name header should be present"); + + // All whitelisted response headers must be returned + Map headers = response.getResponseHeaders(); + Assert.assertNotNull(headers, "response headers map should be present"); + Assert.assertEquals(headers.get(ClickHouseHttpProto.HEADER_QUERY_ID), uuid, + "query id header should round-trip"); + Assert.assertNotNull(headers.get(ClickHouseHttpProto.HEADER_TIMEZONE), + "timezone header should be present"); + Assert.assertNotNull(headers.get(ClickHouseHttpProto.HEADER_FORMAT), + "format header should be present"); + Assert.assertNotNull(headers.get(ClickHouseHttpProto.HEADER_SRV_DISPLAY_NAME), + "server display name header should be present"); + Assert.assertNotNull(headers.get(ClickHouseHttpProto.HEADER_SRV_SUMMARY), + "summary header should be present"); + Assert.assertEquals(headers.get(ClickHouseHttpProto.HEADER_SRV_DISPLAY_NAME), + response.getServerDisplayName(), "getter should match header map value"); + } + } + protected final static List DATASET_COLUMNS = Arrays.asList( "col1 UInt32", "col2 Int32", From 47a49658d9ab1ce32fdfdeef611edbcd20583d38 Mon Sep 17 00:00:00 2001 From: Sergey Chernov Date: Fri, 26 Jun 2026 08:20:33 -0700 Subject: [PATCH 3/3] Added cancelation tests --- .../clickhouse/client/HttpTransportTests.java | 181 ++++++++++++++++++ 1 file changed, 181 insertions(+) diff --git a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java index 6d1a27010..4f7953d81 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java @@ -21,13 +21,19 @@ import com.clickhouse.client.api.insert.InsertResponse; import com.clickhouse.client.api.insert.InsertSettings; import com.clickhouse.client.api.internal.DataTypeConverter; +import com.clickhouse.client.api.internal.HttpAPIClientHelper; import com.clickhouse.client.api.internal.ServerSettings; import com.clickhouse.client.api.internal.ValidationUtils; import com.clickhouse.client.api.query.GenericRecord; import com.clickhouse.client.api.query.QueryResponse; import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.client.api.transport.Endpoint; +import com.clickhouse.client.api.transport.HttpEndpoint; +import com.clickhouse.client.api.transport.internal.TransportRequest; +import com.clickhouse.client.api.transport.internal.TransportResponse; import com.clickhouse.client.config.ClickHouseClientOption; import com.clickhouse.data.ClickHouseFormat; +import net.jpountz.lz4.LZ4Factory; import com.github.tomakehurst.wiremock.WireMockServer; import com.github.tomakehurst.wiremock.client.WireMock; import com.github.tomakehurst.wiremock.common.ConsoleNotifier; @@ -64,6 +70,7 @@ import javax.net.ssl.SSLHandshakeException; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.InputStream; import java.io.StringWriter; import java.math.BigInteger; import java.net.InetAddress; @@ -102,11 +109,16 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.zip.GZIPOutputStream; import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED; +import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @Test(groups = {"integration"}) @@ -2532,4 +2544,173 @@ private static String toPem(X509Certificate certificate) throws Exception { } return stringWriter.toString(); } + + /** + * Exercises {@code HttpAPIClientHelper.TransportRequestImpl#cancel()}, which is not reachable through the + * high-level {@link Client} API. A {@link HttpAPIClientHelper} is created directly and used to start a request + * for an effectively endless stream (reading from {@code system.numbers}) to emulate a long-running query. + * The test verifies the whole life cycle against the server: + *
    + *
  1. the query is observed running on the server ({@code system.processes});
    2. + *
  2. the in-flight request is cancelled from another thread and the reader stops with an error;
    3. + *
  3. the query is no longer running on the server;
    4. + *
  4. {@code system.query_log} contains a record proving the query was interrupted (not finished).
    5. + *
+ */ + @Test(groups = {"integration"}) + @SuppressWarnings("java:S2925") + public void testTransportRequestCancel() throws Exception { + if (isCloud()) { + return; // relies on direct transport access and local system tables (processes / query_log) + } + + ClickHouseNode server = getServer(ClickHouseProtocol.HTTP); + String queryId = "transport-cancel-" + UUID.randomUUID(); + + Map configuration = new HashMap<>(); + configuration.put(ClientConfigProperties.USER.getKey(), "default"); + configuration.put(ClientConfigProperties.PASSWORD.getKey(), ClickHouseServerForTest.getPassword()); + configuration.put(ClientConfigProperties.DATABASE.getKey(), ClickHouseServerForTest.getDatabase()); + configuration.put(ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getKey(), Boolean.FALSE); + configuration.put(ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getKey(), Boolean.FALSE); + + HttpAPIClientHelper helper = new HttpAPIClientHelper(new HashMap<>(configuration), null, false, + LZ4Factory.fastestInstance()); + + try (Client verifyClient = new Client.Builder() + .addEndpoint(Protocol.HTTP, server.getHost(), server.getPort(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .compressClientRequest(false) + .compressServerResponse(false) + .build()) { + + Endpoint endpoint = new HttpEndpoint(server.getHost(), server.getPort(), false, "/"); + + Map requestConfig = new HashMap<>(configuration); + requestConfig.put(ClientConfigProperties.INPUT_OUTPUT_FORMAT.getKey(), ClickHouseFormat.TSV); + requestConfig.put(ClientConfigProperties.QUERY_ID.getKey(), queryId); + + // Endless result set so the query stays active on the server until the request is cancelled. + TransportRequest request = helper.createRequest(endpoint, requestConfig, + "SELECT number FROM system.numbers"); + + AtomicBoolean readStarted = new AtomicBoolean(false); + AtomicLong bytesRead = new AtomicLong(0); + AtomicReference readError = new AtomicReference<>(); + + Thread reader = new Thread(() -> { + long start = System.currentTimeMillis(); + try (TransportResponse response = helper.executeRequest(request); + InputStream in = response.createDataInputStream()) { + byte[] buffer = new byte[8192]; + int read; + while ((read = in.read(buffer)) != -1) { + readStarted.set(true); + bytesRead.addAndGet(read); + // Safety valve so the test can never hang on the endless stream if cancel() fails. + if (System.currentTimeMillis() - start > 30_000) { + break; + } + } + } catch (Throwable t) { + readStarted.set(true); + readError.set(t); + } + }, "transport-request-reader"); + reader.start(); + + // 1. The query must actually be running on the server. + Assert.assertTrue(waitForCondition(() -> isQueryRunning(verifyClient, queryId), 20_000), + "query was not observed running on the server (query_id=" + queryId + ")"); + Assert.assertNull(readError.get(), "reading should still be in progress while the query runs"); + + // 2. Cancel the in-flight request from a separate thread. + AtomicBoolean cancelled = new AtomicBoolean(false); + AtomicReference cancelError = new AtomicReference<>(); + Thread canceller = new Thread(() -> { + try { + cancelled.set(request.cancel()); + } catch (Throwable t) { + cancelError.set(t); + } + }, "transport-request-canceller"); + canceller.start(); + canceller.join(5_000); + + Assert.assertNull(cancelError.get(), "cancel() must not throw"); + Assert.assertTrue(cancelled.get(), "cancel() should report the request as cancelled"); + + // The reader must stop with an error once the underlying connection is aborted. + reader.join(15_000); + Assert.assertFalse(reader.isAlive(), "reading must stop after the request was cancelled"); + Assert.assertNotNull(readError.get(), + "reading a cancelled request must fail, but read " + bytesRead.get() + " bytes"); + + // 3. The server must stop running the query shortly after the client disconnects. + Assert.assertTrue(waitForCondition(() -> !isQueryRunning(verifyClient, queryId), 20_000), + "query is still running on the server after cancellation (query_id=" + queryId + ")"); + + // 4. The query_log must show the query was interrupted instead of finishing successfully. + assertQueryWasInterrupted(verifyClient, queryId); + assertTrue(request.cancel()); + } finally { + helper.close(); + } + } + + private static boolean isQueryRunning(Client client, String queryId) { + List rows = client.queryAll( + "SELECT count() AS c FROM system.processes WHERE query_id = '" + queryId + "'"); + return !rows.isEmpty() && rows.get(0).getLong("c") > 0; + } + + private static boolean waitForCondition(Supplier condition, long timeoutMillis) + throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMillis; + while (System.currentTimeMillis() < deadline) { + try { + if (Boolean.TRUE.equals(condition.get())) { + return true; + } + } catch (Exception ignore) { + // transient query failures (e.g. server busy) are retried until the timeout elapses + } + Thread.sleep(100); + } + return false; + } + + private static void assertQueryWasInterrupted(Client client, String queryId) throws InterruptedException { + long deadline = System.currentTimeMillis() + 30_000; + String seenTypes = ""; + while (System.currentTimeMillis() < deadline) { + client.queryAll("SYSTEM FLUSH LOGS"); + List rows = client.queryAll( + "SELECT toString(type) AS type, exception_code FROM system.query_log " + + "WHERE query_id = '" + queryId + "' AND event_date >= today() - 1"); + + StringBuilder types = new StringBuilder(); + for (GenericRecord row : rows) { + String type = row.getString("type"); + int exceptionCode = row.getInteger("exception_code"); + types.append(type).append('(').append(exceptionCode).append(") "); + + Assert.assertNotEquals(type, "QueryFinish", + "query unexpectedly finished successfully instead of being interrupted (query_id=" + + queryId + ")"); + if ("ExceptionWhileProcessing".equals(type)) { + Assert.assertNotEquals(exceptionCode, 0, + "an interrupted query must be logged with a non-zero exception code (query_id=" + + queryId + ")"); + return; // found the proof the query was interrupted on the server + } + } + seenTypes = types.toString(); + Thread.sleep(250); + } + Assert.fail("no query_log record proving the query was interrupted (query_id=" + queryId + + ", seen types: [" + seenTypes + "])"); + } }