Add client-v2 retry cause for the TIMEOUT_EXCEEDED server error

CHANGELOG.md

@@ -24,6 +24,11 @@
   the previous size can restore it by setting both options to 804800.
   (https://github.com/ClickHouse/clickhouse-java/issues/3121)
 
+- **[client-v2, jdbc-v2]** Server error `159` (`TIMEOUT_EXCEEDED`) is not retried by default anymore. It has its own
+  retry cause, `ClientFaultCause.ServerTimeoutExceeded`, which `ServerRetryable` does not include and which is not in
+  the default `client_retry_on_failures` list. Add `ServerTimeoutExceeded` to the list to retry this error.
+  (https://github.com/ClickHouse/clickhouse-java/issues/3072)
+
 - **[r2dbc]** Pre `1.0.0` is not supported anymore because R2DBC API reached stable `1.0.0` version.  
 
 ### New Features

client-v2/src/main/java/com/clickhouse/client/api/ClientFaultCause.java

@@ -9,4 +9,10 @@
     ConnectionRequestTimeout,
     SocketTimeout,
     ServerRetryable,
+
+    /**
+     * Server error {@code 159} ({@code TIMEOUT_EXCEEDED}). Not covered by {@link #ServerRetryable} and not retried
+     * by default.
+     */
+    ServerTimeoutExceeded,
 }

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

@@ -8,6 +8,8 @@ public class ServerException extends ClickHouseException {
 
     public static final int UNKNOWN_SETTING = 115;
 
+    public static final int EXECUTION_TIMEOUT = 159;
+
     private final int code;
 
     private final int transportProtocolCode;

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

@@ -1098,7 +1098,7 @@
         return converter.apply(header.getValue());
     }
 
     public boolean shouldRetry(Throwable ex, Map<String, Object> requestSettings) {
         List<ClientFaultCause> retryCauses = ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getOrDefault(requestSettings);
 
         if (retryCauses.contains(ClientFaultCause.None)) {
@@ -1130,6 +1130,9 @@
         // there are some db retryable error codes
         if (ex instanceof ServerException || ex.getCause() instanceof ServerException) {
             ServerException se = (ServerException) (ex instanceof ServerException ? ex : ex.getCause());
+            if (se.getCode() == ServerException.EXECUTION_TIMEOUT) {
+                return retryCauses.contains(ClientFaultCause.ServerTimeoutExceeded);
+            }
             return se.isRetryable() && retryCauses.contains(ClientFaultCause.ServerRetryable);
         }
 

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

@@ -1,6 +1,7 @@
 package com.clickhouse.client.api.internal;
 
 import com.clickhouse.client.api.ClientConfigProperties;
+import com.clickhouse.client.api.ClientFaultCause;
 import com.clickhouse.client.api.ServerException;
 import com.clickhouse.client.api.enums.SSLMode;
 import com.clickhouse.client.api.http.ClickHouseHttpProto;
@@ -341,8 +342,7 @@ public void testExecuteRequestThrowsConnectExceptionOn503() throws Exception {
 
     @DataProvider(name = "serverExceptionRetryCases")
     public static Object[][] serverExceptionRetryCases() {
-        // Server code 159 (TIMEOUT_EXCEEDED) is retryable; code 60 (TABLE_NOT_FOUND) is not.
-        ServerException retryable = new ServerException(159, "TIMEOUT_EXCEEDED", 500, "q1");
+        ServerException retryable = new ServerException(202, "TOO_MANY_SIMULTANEOUS_QUERIES", 500, "q1");
         ServerException nonRetryable = new ServerException(60, "TABLE_NOT_FOUND", 404, "q2");
         return new Object[][]{
                 // ServerException thrown directly (behaviour that already worked; pinned as contrast).
@@ -368,6 +368,36 @@ public void testShouldRetryUsesServerExceptionFromCause(Throwable ex, boolean ex
         assertEquals(helper.shouldRetry(ex, new HashMap<>()), expectedRetry);
     }
 
+    @DataProvider(name = "serverTimeoutExceededRetryCases")
+    public static Object[][] serverTimeoutExceededRetryCases() {
+        ServerException timeout = new ServerException(ServerException.EXECUTION_TIMEOUT, "TIMEOUT_EXCEEDED", 500, "q1");
+        ServerException tooManyQueries = new ServerException(202, "TOO_MANY_SIMULTANEOUS_QUERIES", 500, "q2");
+        List<ClientFaultCause> timeoutOnly = Collections.singletonList(ClientFaultCause.ServerTimeoutExceeded);
+        List<ClientFaultCause> retryableOnly = Collections.singletonList(ClientFaultCause.ServerRetryable);
+        return new Object[][]{
+                {null, timeout, false},
+                {timeoutOnly, timeout, true},
+                {timeoutOnly, new RuntimeException("transport failure", timeout), true},
+                {retryableOnly, timeout, false},
+                {retryableOnly, new RuntimeException("transport failure", timeout), false},
+                {timeoutOnly, tooManyQueries, false},
+                {retryableOnly, tooManyQueries, true},
+                {Arrays.asList(ClientFaultCause.ServerRetryable, ClientFaultCause.ServerTimeoutExceeded), timeout, true},
+                {Arrays.asList(ClientFaultCause.None, ClientFaultCause.ServerTimeoutExceeded), timeout, false},
+        };
+    }
+
+    @Test(dataProvider = "serverTimeoutExceededRetryCases")
+    public void testShouldRetryServerTimeoutExceeded(List<ClientFaultCause> retryCauses, Throwable ex,
+                                                     boolean expectedRetry) {
+        HttpAPIClientHelper helper = HttpAPIClientHelperFactory.newHelper(new HashMap<>(), LZ4Factory.fastestInstance());
+        Map<String, Object> requestSettings = new HashMap<>();
+        if (retryCauses != null) {
+            requestSettings.put(ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getKey(), retryCauses);
+        }
+        assertEquals(helper.shouldRetry(ex, requestSettings), expectedRetry);
+    }
+
     /**
      * A multipart body (statement parameters sent as form data) is never compressed, so the request must not
      * declare a content encoding - the server would try to decompress the plain body and fail with

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

@@ -6,6 +6,7 @@
 import com.clickhouse.client.ClickHouseProtocol;
 import com.clickhouse.client.ClickHouseServerForTest;
 import com.clickhouse.client.api.Client;
+import com.clickhouse.client.api.ClientConfigProperties;
 import com.clickhouse.client.api.ClientFaultCause;
 import com.clickhouse.client.api.DataStreamWriter;
 import com.clickhouse.client.api.ServerException;
@@ -21,6 +22,7 @@
 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.ResponseDefinitionBuilder;
 import com.github.tomakehurst.wiremock.client.WireMock;
 import com.github.tomakehurst.wiremock.common.ConsoleNotifier;
 import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
@@ -38,6 +40,7 @@
 import java.io.OutputStream;
 import java.lang.reflect.Field;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -231,6 +234,146 @@
                 "[" + operation + "] retries are disabled, no retry expected");
     }
 
+    @DataProvider(name = "timeoutExceededRetryProvider")
+    public static Object[][] timeoutExceededRetryProvider() {
+        List<Object[]> cases = new ArrayList<>();
+        for (Object[] op : operationProvider()) {
+            cases.add(new Object[]{op[0], op[1], new ClientFaultCause[0], false});
+            cases.add(new Object[]{op[0], op[1], new ClientFaultCause[]{ClientFaultCause.ServerTimeoutExceeded}, true});
+            cases.add(new Object[]{op[0], op[1], new ClientFaultCause[]{ClientFaultCause.ServerRetryable}, false});
+        }
+        return cases.toArray(new Object[0][]);
+    }
+
+    /**
+     * Server error {@code TIMEOUT_EXCEEDED} is retried only when {@link ClientFaultCause#ServerTimeoutExceeded}
+     * is among the retry causes. It is not in the default list and {@link ClientFaultCause#ServerRetryable}
+     * does not cover it.
+     */
+    @Test(groups = {"integration"}, dataProvider = "timeoutExceededRetryProvider")
+    public void testTimeoutExceededRetriedOnlyForItsFaultCause(String operation,
+                                                               ThrowingFunction<Client, Void> function,
+                                                               ClientFaultCause[] retryOn, boolean expectRetry) {
+        if (isCloud()) {
+            return; // mocked server
+        }
+
+        WireMockServer mockServer = startMockServer();
+        mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
+                .inScenario("Timeout")
+                .whenScenarioStateIs(STARTED)
+                .willSetStateTo("Recovered")
+                .willReturn(timeoutExceededResponse()).build());
+        mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
+                .inScenario("Timeout")
+                .whenScenarioStateIs("Recovered")
+                .willReturn(WireMock.aResponse()
+                        .withStatus(HttpStatus.SC_OK)
+                        .withHeader("X-ClickHouse-Summary",
+                                "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")).build());
+
+        Integer errorCode = null;
+        try (Client client = mockServerClient(mockServer, 1, retryOn)) {
+            function.apply(client);
+        } catch (ServerException e) {
+            errorCode = e.getCode();
+        } catch (Exception e) {
+            Assert.fail("[" + operation + "] unexpected exception type", e);
+        } finally {
+            mockServer.stop();
+        }
+
+        Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(),
+                expectRetry ? 2 : 1, "[" + operation + "] unexpected number of attempts");
+        Assert.assertEquals(errorCode, expectRetry ? null : ServerException.EXECUTION_TIMEOUT,
+                "[" + operation + "] unexpected outcome");
+    }
+
+    @DataProvider(name = "perOperationTimeoutRetryProvider")
+    public static Object[][] perOperationTimeoutRetryProvider() {
+        ClientFaultCause[] clientDefault = new ClientFaultCause[0];
+        ClientFaultCause[] clientWithTimeout = {ClientFaultCause.ServerRetryable, ClientFaultCause.ServerTimeoutExceeded};
+        List<Object[]> cases = new ArrayList<>();
+        for (Object[] op : operationsWithRetryCauses(Collections.singletonList(ClientFaultCause.ServerTimeoutExceeded))) {
+            cases.add(new Object[]{op[0], op[1], clientDefault, true});
+        }
+        for (Object[] op : operationsWithRetryCauses(Collections.singletonList(ClientFaultCause.ServerRetryable))) {
+            cases.add(new Object[]{op[0], op[1], clientWithTimeout, false});
+        }
+        return cases.toArray(new Object[0][]);
+    }
+
+    private static Object[][] operationsWithRetryCauses(List<ClientFaultCause> retryCauses) {
+        String key = ClientConfigProperties.CLIENT_RETRY_ON_FAILURE.getKey();
+        ThrowingFunction<Client, Void> query = (client) -> {
+            try (QueryResponse response = client.query("SELECT 1",
+                    new QuerySettings().setOption(key, retryCauses)).get(30, TimeUnit.SECONDS)) {
+                return null;
+            }
+        };
+        ThrowingFunction<Client, Void> streamInsert = (client) -> {
+            try (InsertResponse response = client.insert("table01",
+                    new ByteArrayInputStream("1\t2\t3\n".getBytes()), ClickHouseFormat.TSV,
+                    new InsertSettings().setOption(key, retryCauses)).get(30, TimeUnit.SECONDS)) {
+                return null;
+            }
+        };
+        ThrowingFunction<Client, Void> pojoInsert = (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)),
+                    new InsertSettings().setOption(key, retryCauses)).get(30, TimeUnit.SECONDS)) {
+                return null;
+            }
+        };
+        return new Object[][]{
+                {"query", query},
+                {"insert-stream", streamInsert},
+                {"insert-pojo", pojoInsert}
+        };
+    }
+
+    /**
+     * Retry causes in operation settings replace the client ones, so one operation can enable or disable the retry
+     * of {@code TIMEOUT_EXCEEDED} independently of the client configuration.
+     */
+    @Test(groups = {"integration"}, dataProvider = "perOperationTimeoutRetryProvider")
+    public void testTimeoutExceededRetryCausesPerOperation(String operation,
+                                                           ThrowingFunction<Client, Void> function,
+                                                           ClientFaultCause[] clientRetryOn, boolean expectRetry) {
+        if (isCloud()) {
+            return; // mocked server
+        }
+
+        WireMockServer mockServer = startMockServer();
+        mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
+                .willReturn(timeoutExceededResponse()).build());
+
+        int maxRetries = 2;
+        try (Client client = mockServerClient(mockServer, maxRetries, clientRetryOn)) {
+            function.apply(client);
+            Assert.fail("[" + operation + "] expected TIMEOUT_EXCEEDED error");
+        } catch (ServerException e) {
+            Assert.assertEquals(e.getCode(), ServerException.EXECUTION_TIMEOUT);
+        } catch (Exception e) {
+            Assert.fail("[" + operation + "] unexpected exception type", e);
+        } finally {
+            mockServer.stop();
+        }
+
+        Assert.assertEquals(mockServer.findAll(WireMock.postRequestedFor(WireMock.anyUrl())).size(),
+                expectRetry ? maxRetries + 1 : 1, "[" + operation + "] unexpected number of attempts");
+    }
+
+    private static ResponseDefinitionBuilder timeoutExceededResponse() {
+        return WireMock.aResponse()
+                .withStatus(HttpStatus.SC_INTERNAL_SERVER_ERROR)
+                .withHeader("X-ClickHouse-Exception-Code", String.valueOf(ServerException.EXECUTION_TIMEOUT))
+                .withBody("Code: 159. DB::Exception: Timeout exceeded: elapsed 1.000 sec., maximum: 0.001 sec. "
+                        + "(TIMEOUT_EXCEEDED)");
+    }
+
     /**
      * When no query id is supplied by the caller but a generator is configured, the operation must
      * use the generated id. Covers the

docs/features.md

@@ -33,7 +33,7 @@ This document lists stable, user-visible behavior in `client-v2` and `jdbc-v2` t
 - Metadata discovery: Loads table schemas from table names or queries and allows schema registration for typed read/write operations.
 - 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.
+- Retry behavior: Can retry failed operations for configured failure causes and retry limits. Server error `159` (`TIMEOUT_EXCEEDED`) is retried only when its own failure cause, `ServerTimeoutExceeded`, is in `client_retry_on_failures`; it is not in the default list and `ServerRetryable` does not cover it. The list can also be set in the settings of one operation, where it replaces the client list.
 - Client-side request cancellation: `Client.cancelTransportRequest(String queryId)` aborts the in-flight HTTP request and its IO for the operation started with the given query id. It requires the caller to set the query id in operation settings, is best-effort (it cancels client-side IO but the result is not guaranteed), and does not issue a server-side `KILL QUERY` - the server stops the query on its own once the client disconnects. A cancelled operation that is being retried stops instead of issuing another request, also when the cancellation lands between two attempts (for example from `DataStreamWriter#onRetry()`).
 - Metrics and observability: Exposes client/server operation metrics and optionally integrates connection-pool gauges with Micrometer.
 - Span recording (tracing SPI): `Client.Builder.setSpanRecorder(SpanRecorder)` registers a backend-agnostic recorder (package `com.clickhouse.client.api.observability`) that observes client operations. Every operation - a query, a command or an insert, including the `ping` and `getTableSchema` calls, which run a query - starts one operation span, and every transport request made for it, including each retry, starts a child request span. `SpanRecorder` and `Span` are plain interfaces; an implementation extends the `DefaultSpanRecorder` base class and overrides only what it wants to record, so a recorder keeps working when the client starts a kind of span it does not know about. The registered recorder is the first thing the client calls, and it is called with everything the client knows about the operation - its settings object, the statement, the target table, the batch size, the endpoint, the metrics of the completed operation and the failure - so an implementation is free to record whatever it needs and in whatever form. Deriving the standard span names and attribute values from those structures is done by `SpanSupport`, which a recorder implementation calls if it wants them; using it is opt-in, and its methods may be overridden to report other values. A recorder opts in through `DefaultSpanRecorder#getSpanSupport()` (or `SpanSupport.DEFAULT`). Span names and attribute keys follow the OpenTelemetry semantic conventions for database and HTTP client spans, the keys are defined by the `SpanAttribute` enum, and the values are derived by `SpanSupport`, so every recorder that uses it reports the same information. An operation span is named `query <database>`, or `insert <database>.<table>` for an insert, and carries `db.system.name`, `db.namespace`, `clickhouse.query_id`, `db.query.text` (query/command), `db.query.parameter.<name>` (parameterized query), `db.operation.name` (insert), `db.collection.name` (insert), `db.operation.batch.size` (POJO insert), `server.address`/`server.port` of the first configured endpoint (each attempt reports its own on the request span), the metrics of the completed operation on success, and `error.type` plus `db.response.status_code` on failure. Success is reported per operation kind, because the metrics that describe a read are not the ones that describe a write: `SpanRecorder#recordQuerySuccess` is called for a read operation and `SpanRecorder#recordInsertSuccess` for an insert, and each records only the metrics of its own kind - a query reports `db.response.returned_rows`, `clickhouse.response.read_rows` and `clickhouse.response.read_bytes`, an insert reports `clickhouse.response.written_rows` and `clickhouse.response.written_bytes`, and both report `clickhouse.query_id`. These values come from the server's progress summary, which is not always available (for example a query reports them with `QuerySettings#waitEndOfQuery(true)`); a metric the server did not report is left out. The same distinction is on the metrics themselves: `OperationMetrics#getOperationType()` returns `OperationType.QUERY` or `OperationType.INSERT`. It reports the kind of the call the application made, not the kind of work the server did, so a command that writes - `INSERT INTO ... SELECT` run through `execute` - is reported as a query. A request span is named `POST` and carries `http.request.method`, the `server.address`/`server.port` of that attempt, `http.response.status_code` once a response is received, and `error.type`/`db.response.status_code` when the attempt fails. An operation span is started on the calling thread, so it joins the caller's ambient trace even when `async_operations` runs the operation on the client's executor, and it is ended when the operation returns its response to the caller - so it covers sending the request and receiving the response head, not streaming the response body afterwards. Each span of an operation that started is ended exactly once, also when the operation failed; an operation that could not be started at all - a closed client, for example - does not report a span. When no recorder is registered nothing is recorded and no span-related work is done.