diff --git a/CHANGELOG.md b/CHANGELOG.md index 320805a55..4feee82cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,8 +27,14 @@ - **[client-v2]** `Client.Builder#useBearerTokenAuth(String)` now stores the bearer token under the `access_token` configuration key (with the `Bearer ` prefix) instead of writing it directly into `http_header_authorization`. The HTTP wire format is unchanged, but the token is no longer observable through `Client#getReadOnlyConfig()` under the `http_header_authorization` key. +- **[client-v2]** HTTP `503 Service Unavailable` responses are now surfaced as a connection-style failure (`java.net.ConnectException`) and are retried by default. Previously a `503` was treated as a server error (`ServerException`) and fell under the `ServerRetryable` fault cause. It has been moved to the `ConnectTimeout` fault cause category so that connectivity/availability failures are handled uniformly with other connection errors. Callers that specifically excluded `ServerRetryable` to avoid retrying `503` should now adjust their `client_retry_on_failures` configuration to exclude `ConnectTimeout` instead. + +- **[client-v2]** Unexpected/unknown HTTP status codes (those the client cannot interpret as a ClickHouse response) now throw a `ClientException` instead of a `ServerException`. Since the client cannot meaningfully handle these responses, they are reported as a client-side error rather than being attributed to the server. + ### New Features +- **[client-v2]** Added `Client#cancelTransportRequest(String queryId)` to cancel an in-flight request that has not yet received a response from the server, identified by the query id supplied in the operation settings. This aborts the request on the client side (cancels the underlying IO operation) but does **not** issue a `KILL QUERY` on the server, so a query that already started executing may continue to run server-side. It is recommended to use operation timeout settings where possible; this API is intended for explicitly aborting a request from the client. + - **[client-v2]** Added `Session` API to encapsulate and manage ClickHouse session settings (`session_id`, `session_check`, `session_timeout`, `session_timezone`) as a reusable object. The `Session` instance can be applied to any request settings using `applyTo()`, and session state can be cleared via `clearSession()`. Additionally, added `resetOption(String)` to `InsertSettings`, `QuerySettings`, and `CommonSettings` to allow removing specific settings. Settings explicitly set to `null` will not be sent to the server, which is useful for overriding global settings. - **[client-v2]** Added runtime credential update APIs on `Client`: `updateUserAndPassword(String, String)`, `updateAccessToken(String)`, and `updateBearerToken(String)`. Subsequent requests on the same `Client` instance use the new credentials without rebuilding the client. The authentication method is fixed at construction time; calling a runtime updater that does not match the configured method throws `ClientMisconfigurationException`. See `docs/authentication.md` for details and migration guidance. 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 5cf21a52e..56c8aa306 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 @@ -17,6 +17,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; @@ -37,6 +38,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; @@ -44,9 +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.ClassicHttpResponse; -import org.apache.hc.core5.http.Header; -import org.apache.hc.core5.http.HttpStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,6 +65,7 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.StringJoiner; import java.util.TimeZone; @@ -113,6 +114,8 @@ public class Client implements AutoCloseable { private static final Logger LOG = LoggerFactory.getLogger(Client.class); + private static final int REQ_REGISTRY_SIZE = 1000; // initial capacity hint for the ongoing-request registry, not a hard limit; no separate config needed. + private HttpAPIClientHelper httpClientHelper = null; private final List endpoints; @@ -135,6 +138,8 @@ public class Client implements AutoCloseable { private final Map> typeHintMapping; + private final ConcurrentHashMap ongoingRequests = new ConcurrentHashMap<>(REQ_REGISTRY_SIZE); + // Server context private String dbUser; private String serverVersion; @@ -232,7 +237,6 @@ public String getDefaultDatabase() { return (String) this.configuration.get(ClientConfigProperties.DATABASE.getKey()); } - /** * Frees the resources associated with the client. *
    @@ -1374,65 +1378,57 @@ public CompletableFuture insert(String tableName, List data, long startTime = System.nanoTime(); // Selecting some node Endpoint selectedEndpoint = getNextAliveNode(); - + final String queryId = requestSettings.getQueryId(); 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(); + }); + registerTransportReq(queryId, transportRequest); - // Check response - if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), durationSince(startTime)); - selectedEndpoint = getNextAliveNode(); - continue; - } - + try (TransportResponse transportResponse = httpClientHelper.executeRequest(transportRequest)) { 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(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())) { + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(queryId)) { LOG.warn("Retrying.", e); selectedEndpoint = getNextAliveNode(); } else { throw lastException; } + } finally { + unregisterTransportReq(queryId); } } 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()); - } /** @@ -1568,7 +1564,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()) : @@ -1592,55 +1588,54 @@ 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(); 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(); - })) { - - - // Check response - if (httpResponse.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) { - LOG.warn("Failed to get response. Server returned {}. Retrying. (Duration: {})", httpResponse.getCode(), 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)); - } 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())) { - LOG.warn("Retrying.", e); - selectedEndpoint = getNextAliveNode(); - } else { - throw lastException; + final String queryId = requestSettings.getQueryId(); + try { + for (int i = 0; i <= retries; i++) { + // Execute request + TransportRequest transportRequest = httpClientHelper.createRequest(selectedEndpoint, requestSettings.getAllSettings(), + out -> { + writer.onOutput(out); + out.close(); + }); + registerTransportReq(queryId, transportRequest); + + 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()); + if (httpClientHelper.shouldRetry(e, requestSettings.getAllSettings()) && requestIsNotCancelled(requestSettings.getQueryId())) { + LOG.warn("Retrying.", e); + selectedEndpoint = getNextAliveNode(); + } 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 < retries) { - try { - writer.onRetry(); - } catch (IOException ioe) { - throw new ClientException("Failed to reset stream before next attempt", ioe); + if (i < retries) { + try { + writer.onRetry(); + } catch (IOException ioe) { + throw new ClientException("Failed to reset stream before next attempt", ioe); + } } } + } finally { + unregisterTransportReq(queryId); } + String errMsg = requestExMsg("Insert", retries, durationSince(startTime).toMillis(), requestSettings.getQueryId()); LOG.warn(errMsg); throw (lastException == null ? new ClientException(errMsg) : lastException); @@ -1727,48 +1722,37 @@ public CompletableFuture query(String sqlQuery, Map query(String sqlQuery, Map query(String sqlQuery, Map queryParams) { return query(sqlQuery, queryParams, null); } + private OperationMetrics completeOperation(TransportResponse transportResponse, ClientStatisticsHolder clientStats, String originalQueryId) { + OperationMetrics metrics = new OperationMetrics(clientStats); + String summary = transportResponse.getSummaryJson(); + ProcessParser.parseSummary(summary, metrics); + String queryId = transportResponse.getQueryId(); + metrics.setQueryId(queryId == null ? originalQueryId : queryId); + metrics.operationComplete(); + return metrics; + } + /** *

    Queries 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 @@ -2276,6 +2291,23 @@ public void updateAccessToken(String accessToken) { this.credentialsManager.setAccessToken(accessToken); } + /** + * Note: It is recommended to use operation timeout settings instead of this method. + * 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. + * + * @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(); + } + } + private Endpoint getNextAliveNode() { return endpoints.get(0); } 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..9635fc4d7 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,9 @@ package com.clickhouse.client.api.internal; +import org.slf4j.Logger; + +import java.io.Closeable; + /** * Class containing utility methods used across the client. */ @@ -14,4 +18,14 @@ public static boolean isNotBlank(String str) { public static boolean isBlank(String str) { return str == null || str.trim().isEmpty(); } + + public static void quietClose(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 ab4b0153c..6e3820dde 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 @@ -15,6 +15,8 @@ import com.clickhouse.client.api.enums.SSLMode; 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.data.ClickHouseFormat; import net.jpountz.lz4.LZ4Factory; import org.apache.commons.compress.compressors.CompressorStreamFactory; @@ -25,6 +27,7 @@ import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder; import org.apache.hc.client5.http.impl.classic.CloseableHttpClient; import org.apache.hc.client5.http.impl.classic.HttpClientBuilder; +import org.apache.hc.client5.http.impl.classic.RequestFailedException; import org.apache.hc.client5.http.impl.io.BasicHttpClientConnectionManager; import org.apache.hc.client5.http.impl.io.ManagedHttpClientConnectionFactory; import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager; @@ -43,8 +46,10 @@ import org.apache.hc.core5.http.HttpHeaders; import org.apache.hc.core5.http.HttpHost; import org.apache.hc.core5.http.HttpRequest; +import org.apache.hc.core5.http.HttpResponse; 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; @@ -99,6 +104,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; import java.util.function.Function; @@ -365,10 +371,7 @@ public CloseableHttpClient createHttpClient(boolean initSslContext, Map 0 ? readClickHouseError(httpResponse.getEntity(), serverCode, queryId, httpResponse.getCode()) : @@ -529,9 +532,44 @@ 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; + private final AtomicBoolean cancelled = new AtomicBoolean(false); + + TransportRequestImpl(HttpPost delegate, Map config) { + this.delegate = delegate; + this.config = config; + } + + @Override + public boolean cancel() { + cancelled.set(true); + if (delegate.isCancelled()) { + return true; + } + return delegate.cancel(); + } + + @Override + public boolean isCancelled() { + return cancelled.get(); + } + + @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); @@ -554,34 +592,81 @@ 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 { - 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)); + private static final class TransportResponseImpl implements TransportResponse { + + private final ClassicHttpResponse delegate; + + TransportResponseImpl(ClassicHttpResponse delegate) { + this.delegate = delegate; + } - return doPostRequest(requestConfig, req); + @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 + @SuppressWarnings("unchecked") + 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); + } + } } - private ClassicHttpResponse doPostRequest(Map requestConfig, HttpPost req) throws Exception { + public TransportResponse executeRequest(TransportRequest transportRequest) throws Exception { + + final Map requestConfig = transportRequest.getConfig(); + final HttpPost req = transportRequest.getDelegate(); doPoolVent(); ClassicHttpResponse httpResponse = null; + boolean closeResponse = true; HttpContext context = createRequestHttpContext(requestConfig); try { httpResponse = httpClient.executeOpen(null, req, context); @@ -590,43 +675,73 @@ private ClassicHttpResponse doPostRequest(Map requestConfig, Htt httpResponse.getCode(), requestConfig)); - if (httpResponse.getCode() == HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED) { - throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings."); - } else if (httpResponse.getCode() == HttpStatus.SC_BAD_GATEWAY) { - httpResponse.close(); - throw new ClientException("Server returned '502 Bad gateway'. Check network and proxy settings."); - } else if (httpResponse.getCode() >= HttpStatus.SC_BAD_REQUEST || httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) { - try { - throw readError(req, httpResponse); - } finally { - httpResponse.close(); - } + if (httpResponse.containsHeader(ClickHouseHttpProto.HEADER_EXCEPTION_CODE)) { + throw readError(req, httpResponse); } - return httpResponse; + int statusCode = httpResponse.getCode(); + switch (statusCode) { + case HttpStatus.SC_OK: + closeResponse = false; + return new TransportResponseImpl(httpResponse); + case HttpStatus.SC_PROXY_AUTHENTICATION_REQUIRED: + throw new ClientMisconfigurationException("Proxy authentication required. Please check your proxy settings."); + case HttpStatus.SC_BAD_GATEWAY: + throw new ClientException("Server returned '502 Bad gateway'. Check network and proxy settings."); + case HttpStatus.SC_SERVICE_UNAVAILABLE: + throw new ConnectException("Server returned '503 Service Unavailable'. Check network settings."); + case HttpStatus.SC_BAD_REQUEST: + case HttpStatus.SC_UNAUTHORIZED: + case HttpStatus.SC_FORBIDDEN: + case HttpStatus.SC_SERVER_ERROR: + case HttpStatus.SC_NOT_FOUND: + // ClickHouse usually uses SC_BAD_REQUEST and SC_SERVER_ERROR to return error. + // SC_UNAUTHORIZED, SC_FORBIDDEN is for authentication + // SC_NOT_FOUND can be returned by ClickHouse when path doesn't match database, but also by proxy + // others we cannot handle properly + throw readError(req, httpResponse); + default: + throw new ClientException("Unexpected result status " + statusCode); + } } catch (UnknownHostException e) { - closeQuietly(httpResponse); LOG.warn("Host '{}' unknown", req.getAuthority()); throw e; } catch (ConnectException | NoRouteToHostException e) { - closeQuietly(httpResponse); LOG.warn("Failed to connect to '{}': {}", req.getAuthority(), e.getMessage()); throw e; } catch (Exception e) { - closeQuietly(httpResponse); LOG.debug("Failed to execute request to '{}': {}", req.getAuthority(), e.getMessage(), e); + if (e instanceof RequestFailedException && req.isCancelled()) { + throw new TransportException("Request was cancelled on client side", e, getQueryId(httpResponse, req)); + } throw e; + } finally { + if (closeResponse) { + ClientUtils.quietClose(httpResponse, LOG); + } } } - public static void closeQuietly(ClassicHttpResponse httpResponse) { - if (httpResponse != null) { - try { - httpResponse.close(); - } catch (IOException e) { - LOG.warn("Failed to close response"); - } + public TransportRequest createRequest(Endpoint server, Map requestConfig, IOCallback writeCallback) { + final URI uri = createRequestURI(server, requestConfig, true); + final HttpPost req = createPostRequest(uri, 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 new TransportRequestImpl(req, requestConfig); + } + + private String getQueryId(HttpResponse httpResponse, HttpPost httpRequest) { + final Header serverQueryIdHeader = httpResponse == null ? null : httpResponse.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); + final Header clientQueryIdHeader = httpRequest == null ? null : httpRequest.getFirstHeader(ClickHouseHttpProto.HEADER_QUERY_ID); + final Header queryHeader = Stream.of(serverQueryIdHeader, clientQueryIdHeader).filter(Objects::nonNull).findFirst().orElse(null); + return queryHeader == null ? "" : queryHeader.getValue(); } private static final ContentType CONTENT_TYPE = ContentType.create(ContentType.TEXT_PLAIN.getMimeType(), "UTF-8"); @@ -820,7 +935,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 )); /** @@ -893,8 +1011,11 @@ public boolean shouldRetry(Throwable ex, Map requestSettings) { // This method wraps some client specific exceptions into specific ClientException or just ClientException // ClientException will be also wrapped public RuntimeException wrapException(String message, Exception cause, String queryId) { - if (cause instanceof ClientException || cause instanceof ServerException) { - return (RuntimeException) cause; + // Already-classified exceptions (ClientException, ServerException, ConnectionInitiationException, ...) + // are returned as-is so their specific type is preserved instead of being reboxed as a generic + // ClickHouseException. + if (cause instanceof ClickHouseException) { + return (ClickHouseException) cause; } if (cause instanceof SSLException) { 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..595cf54b5 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 @@ -5,13 +5,12 @@ 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 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.util.Map; +import java.util.Objects; import java.util.TimeZone; /** @@ -31,54 +30,46 @@ 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, + public QueryResponse(TransportResponse 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; + Objects.requireNonNull(response, "response is null"); + 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 ) { - try { - httpResponse.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/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..79f656425 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportRequest.java @@ -0,0 +1,41 @@ +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. + */ + boolean cancel(); + + /** + * Returns indication if request was canceled on client side + * @return true if request was canceled + */ + boolean isCancelled(); +} 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..beecea34d --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/transport/internal/TransportResponse.java @@ -0,0 +1,44 @@ +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 { + + /** + * 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(); +} 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..cd62bb95b 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"}) @@ -2102,6 +2114,101 @@ public static Object[][] testHttpStatusCompressedBodyDataProvider() { }; } + @Test(groups = {"integration"}) + public void test503ServiceUnavailableSurfacesAsConnectionInitiationException() throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = new WireMockServer(WireMockConfiguration + .options().dynamicPort().notifier(new ConsoleNotifier(false))); + mockServer.start(); + + try { + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withBody("Service Unavailable")) + .build()); + + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressServerResponse(false) + .retryOnFailures(ClientFaultCause.None) + .build()) { + + Throwable thrown = Assert.expectThrows(Throwable.class, + () -> client.query("SELECT 1").get(10, TimeUnit.SECONDS)); + + ConnectionInitiationException connectionInitiationException = + findCause(thrown, ConnectionInitiationException.class); + Assert.assertNotNull(connectionInitiationException, + "Expected 503 to be reported as a ConnectionInitiationException, but was: " + thrown); + + Assert.assertTrue(containsMessageInCauseChain(thrown, "503 Service Unavailable"), + "Expected '503 Service Unavailable' in failure message chain, but was: " + thrown); + + Assert.assertNull(findServerException(thrown), + "A bare 503 (no ClickHouse exception code) should not be reported as a ServerException, but was: " + thrown); + } + } finally { + mockServer.stop(); + } + } + + @Test(groups = {"integration"}) + public void test503ServiceUnavailableIsRetried() throws Exception { + if (isCloud()) { + return; // mocked server + } + + WireMockServer mockServer = new WireMockServer(WireMockConfiguration + .options().dynamicPort().notifier(new ConsoleNotifier(false))); + mockServer.start(); + + try { + // First request fails with a retryable 503 server error + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("ServiceUnavailable") + .withRequestBody(WireMock.containing("SELECT 1")) + .whenScenarioStateIs(STARTED) + .willSetStateTo("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_SERVICE_UNAVAILABLE) + .withHeader("X-ClickHouse-Exception-Code", "202") + .withBody("Code: 202. DB::Exception: Too many simultaneous queries. (TOO_MANY_SIMULTANEOUS_QUERIES)")) + .build()); + + // Second request (retry) succeeds + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .inScenario("ServiceUnavailable") + .withRequestBody(WireMock.containing("SELECT 1")) + .whenScenarioStateIs("Recovered") + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withHeader("X-ClickHouse-Summary", "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")) + .build()); + + try (Client client = new Client.Builder().addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressServerResponse(false) + .setMaxRetries(1) + .retryOnFailures(ClientFaultCause.ServerRetryable) + .build()) { + + try (QueryResponse response = client.query("SELECT 1").get(10, TimeUnit.SECONDS)) { + Assert.assertNotNull(response); + } + } + + mockServer.verify(2, WireMock.postRequestedFor(WireMock.anyUrl())); + } finally { + mockServer.stop(); + } + } + private byte[] fetchBinaryPayload(String query, int networkBufferSize, int timeoutSec) throws Exception { QuerySettings querySettings = new QuerySettings().setFormat(ClickHouseFormat.RowBinaryWithNamesAndTypes); try (Client client = newClient() @@ -2532,4 +2639,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 + "])"); + } } 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..88c5d8526 --- /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 testQuietCloseSwallowsExceptionAndLogs() 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.quietClose(closeable, log); + + Mockito.verify(closeable).close(); + Mockito.verify(log).warn(Mockito.contains("Failed to close object"), Mockito.eq(failure)); + } + + @Test(groups = {"unit"}) + public void testQuietCloseClosesSuccessfully() throws IOException { + Logger log = Mockito.mock(Logger.class); + Closeable closeable = Mockito.mock(Closeable.class); + + ClientUtils.quietClose(closeable, log); + + Mockito.verify(closeable).close(); + Mockito.verifyNoInteractions(log); + } + + @Test(groups = {"unit"}) + public void testQuietCloseWithNull() { + Logger log = Mockito.mock(Logger.class); + + // Should be a no-op and not throw on a null closeable + ClientUtils.quietClose(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..aa299e982 --- /dev/null +++ b/client-v2/src/test/java/com/clickhouse/client/api/transport/TransportBaseTests.java @@ -0,0 +1,612 @@ +package com.clickhouse.client.api.transport; + + +import com.clickhouse.client.BaseIntegrationTest; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +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.insert.InsertSettings; +import com.clickhouse.client.api.metadata.TableSchema; +import com.clickhouse.client.api.query.GenericRecord; +import com.clickhouse.client.api.query.QueryResponse; +import com.clickhouse.client.api.query.QuerySettings; +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.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +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} + }; + } + + /** + * Exercises {@link Client#cancelTransportRequest(String)} for every combination of + * {@code {query, insert} x {sync, async}} operations. A long-running operation is started against the + * real server with a known {@code query_id} and is interrupted from the test thread while it is still + * in progress. The whole life cycle is verified: + *
      + *
    1. the operation is observed running on the server ({@code system.processes});
      2. + *
    2. {@link Client#cancelTransportRequest(String)} aborts the in-flight request and the operation fails;
      3. + *
    3. the query is no longer running on the server.
      4. + *
    + */ + @Test(groups = {"integration"}, dataProvider = "cancelRequestProvider") + @SuppressWarnings("java:S2925") + public void testCancelRequest(String name, boolean async, boolean isInsert) throws Exception { + if (isCloud()) { + return; // relies on local system tables (processes / query_log) + } + + ClickHouseNode server = getServer(ClickHouseProtocol.HTTP); + String queryId = "client-cancel-" + UUID.randomUUID(); + // Fully qualified so the table created via runQuery (default database) matches the one the client + // inserts into (client default database is the test database). + String table = ClickHouseServerForTest.getDatabase() + ".client_cancel_" + + UUID.randomUUID().toString().replace('-', '_'); + + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, server.getHost(), server.getPort(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .setDefaultDatabase(ClickHouseServerForTest.getDatabase()) + .compressClientRequest(false) + .compressServerResponse(false) + .useAsyncRequests(async) + .build()) { + + AtomicReference opError = new AtomicReference<>(); + AtomicBoolean opFinished = new AtomicBoolean(false); + + Runnable operation; + if (isInsert) { + Assert.assertTrue(runQuery("CREATE TABLE " + table + + " (number UInt64) ENGINE = MergeTree ORDER BY number"), "[" + name + "] failed to create table"); + operation = () -> { + // Endless input stream so the insert stays active on the server until the request is cancelled. + try (InsertResponse response = client.insert(table, endlessTsvStream(), ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(35, TimeUnit.SECONDS)) { + opFinished.set(true); + } catch (Throwable t) { + t.printStackTrace(); + opError.set(t); + } + }; + } else { + operation = () -> { + // A long-running aggregation that produces no output until it finishes: the server sends no + // response while it runs, so the worker stays blocked in query().get() waiting for the + // response. That IO wait is exactly what cancelTransportRequest is meant to unblock (once the + // response has been received the request is no longer registered / cancellable). + try (QueryResponse response = client.query( + "SELECT sum(sleepEachRow(1)) FROM numbers(3600) SETTINGS max_block_size = 1", + new QuerySettings().setQueryId(queryId)).get(35, TimeUnit.SECONDS)) { + opFinished.set(true); + } catch (Throwable t) { + opError.set(t); + } + }; + } + + Thread worker = new Thread(operation, "client-cancel-" + name); + worker.start(); + + // 1. The operation must actually be running on the server. + Assert.assertTrue(waitForCondition(() -> isQueryRunning(client, queryId), 20_000), + "[" + name + "] operation was not observed running on the server (query_id=" + queryId + + ", opError=" + opError.get() + ")"); + Assert.assertNull(opError.get(), "[" + name + "] operation should still be in progress while it runs"); + + // 2. Cancel the in-flight request through the high-level Client API. Cancellation is reissued in a + // short loop so it reliably lands while the operation is still in progress, independent of thread + // scheduling races between the worker and the canceller. + cancelUntilStopped(client, queryId, worker, 20_000); + Assert.assertFalse(worker.isAlive(), "[" + name + "] operation must stop after the request was cancelled"); + Assert.assertFalse(opFinished.get(), "[" + name + "] a cancelled operation must not complete successfully"); + Assert.assertNotNull(opError.get(), "[" + name + "] a cancelled operation must fail with an error"); + } finally { + runQuery("DROP TABLE IF EXISTS " + table); + } + } + + @DataProvider(name = "cancelRequestProvider") + public static Object[][] cancelRequestProvider() { + return new Object[][]{ + {"query-sync", false, false}, + {"query-async", true, false}, + {"insert-sync", false, true}, + {"insert-async", true, true} + }; + } + + /** + * A retryable transport failure - here a socket read timeout produced by a mocked server that delays its + * response well beyond the client socket timeout - is normally retried up to {@code maxRetries} times. + * When the caller cancels the in-flight request through {@link Client#cancelTransportRequest(String)} while + * the operation is inside that retry loop, the loop must stop early instead of exhausting the configured + * retries. Covers the {@code requestIsNotCancelled(queryId)} guard of the operation retry loop: the + * operation fails and the mocked server observes fewer than {@code maxRetries + 1} attempts. + */ + @Test(groups = {"integration"}, dataProvider = "cancelDuringRetryProvider") + @SuppressWarnings("java:S2925") + public void testCancelStopsRetriesOnRetryableDelay(String operation, boolean isInsert, boolean isPojo) + throws Exception { + if (isCloud()) { + return; // mocked server + } + + int maxRetries = 50; + WireMockServer mockServer = startMockServer(); + // The response is delayed far beyond the client socket timeout, so every attempt fails with a + // retryable SocketTimeoutException and the operation stays in the retry loop until it is cancelled. + mockServer.addStubMapping(WireMock.post(WireMock.anyUrl()) + .willReturn(WireMock.aResponse() + .withStatus(HttpStatus.SC_OK) + .withFixedDelay(5_000) + .withHeader("X-ClickHouse-Summary", + "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build()); + + String queryId = "cancel-retry-" + UUID.randomUUID(); + AtomicReference opError = new AtomicReference<>(); + AtomicBoolean opFinished = new AtomicBoolean(false); + + try (Client client = new Client.Builder() + .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false) + .setUsername("default") + .setPassword(ClickHouseServerForTest.getPassword()) + .compressClientRequest(false) + .compressServerResponse(false) + .setSocketTimeout(500) + .setMaxRetries(maxRetries) + .retryOnFailures(ClientFaultCause.SocketTimeout) + .build()) { + + Runnable op; + if (isInsert && isPojo) { + client.register(InsertablePojo.class, new TableSchema("table01", null, "default", + Collections.singletonList(ClickHouseColumn.of("id", "Int32")))); + op = () -> { + try (InsertResponse response = client.insert("table01", + Collections.singletonList(new InsertablePojo(1)), + new InsertSettings().setQueryId(queryId)).get(60, TimeUnit.SECONDS)) { + opFinished.set(true); + } catch (Throwable t) { + opError.set(t); + } + }; + } else if (isInsert) { + op = () -> { + try (InsertResponse response = client.insert("table01", + new ByteArrayInputStream("1\t2\t3\n".getBytes()), ClickHouseFormat.TSV, + new InsertSettings().setQueryId(queryId)).get(60, TimeUnit.SECONDS)) { + opFinished.set(true); + } catch (Throwable t) { + opError.set(t); + } + }; + } else { + op = () -> { + try (QueryResponse response = client.query("SELECT timezone()", + new QuerySettings().setQueryId(queryId)).get(60, TimeUnit.SECONDS)) { + opFinished.set(true); + } catch (Throwable t) { + opError.set(t); + } + }; + } + + Thread worker = new Thread(op, "cancel-retry-" + operation); + worker.start(); + + // Cancellation is reissued until the worker stops so it reliably lands while a request is in flight, + // independent of scheduling races between the worker and the canceller. + cancelUntilStopped(client, queryId, worker, 30_000); + + Assert.assertFalse(worker.isAlive(), + "[" + operation + "] operation must stop once the request is cancelled"); + Assert.assertFalse(opFinished.get(), + "[" + operation + "] a cancelled operation must not complete successfully"); + Assert.assertNotNull(opError.get(), + "[" + operation + "] a cancelled operation must fail with an error"); + + int attempts = mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(); + Assert.assertTrue(attempts < maxRetries + 1, + "[" + operation + "] cancellation must stop the retry loop early, but observed " + attempts + + " attempts (maxRetries=" + maxRetries + ")"); + } finally { + mockServer.stop(); + } + } + + @DataProvider(name = "cancelDuringRetryProvider") + public static Object[][] cancelDuringRetryProvider() { + return new Object[][]{ + {"query", false, false}, + {"insert-stream", true, false}, + {"insert-pojo", true, true} + }; + } + + /** + * 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 + * the worker has actually started blocking on the request. + */ + private static void cancelUntilStopped(Client client, String queryId, Thread worker, long timeoutMillis) + throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMillis; + do { + client.cancelTransportRequest(queryId); + worker.join(200); + } while (worker.isAlive() && System.currentTimeMillis() < deadline); + } + + /** + * Produces an effectively endless stream of well-formed single-column TSV rows in 64 KiB chunks. The chunk + * size is large enough to flush through the client's network buffer so the insert is actually observed + * running on the server, while the short sleep keeps the request in progress without flooding it. + */ + private static InputStream endlessTsvStream() { + final byte[] row = "1\n".getBytes(StandardCharsets.US_ASCII); + final byte[] chunk = new byte[64 * 1024]; + for (int i = 0; i < chunk.length; i++) { + chunk[i] = row[i % row.length]; + } + return new InputStream() { + @Override + public int read() { + byte[] single = new byte[1]; + return read(single, 0, 1) == -1 ? -1 : (single[0] & 0xFF); + } + + @Override + public int read(byte[] b, int off, int len) { + try { + Thread.sleep(20); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return -1; + } + int n = Math.min(len, chunk.length); + System.arraycopy(chunk, 0, b, off, n); + return n; + } + }; + } + + 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; + } + + 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", diff --git a/docs/features.md b/docs/features.md index 11a0e4ac4..522781086 100644 --- a/docs/features.md +++ b/docs/features.md @@ -27,6 +27,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. - 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.