diff --git a/client-v2/src/main/java/com/clickhouse/client/api/ClickHouseException.java b/client-v2/src/main/java/com/clickhouse/client/api/ClickHouseException.java new file mode 100644 index 000000000..83da17938 --- /dev/null +++ b/client-v2/src/main/java/com/clickhouse/client/api/ClickHouseException.java @@ -0,0 +1,18 @@ +package com.clickhouse.client.api; + +public class ClickHouseException extends RuntimeException { + protected boolean isRetryable = false; + + public ClickHouseException(String message) { + super(message); + } + + public ClickHouseException(String message, Throwable cause) { + super(message, cause); + } + + public ClickHouseException(Throwable cause) { + super(cause); + } + public boolean isRetryable() { return isRetryable; } +} 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 b014dc99f..cc7890fd2 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 @@ -813,7 +813,7 @@ public Builder setClientNetworkBufferSize(int size) { /** * Sets list of causes that should be retried on. - * Default {@code [NoHttpResponse, ConnectTimeout, ConnectionRequestTimeout]} + * Default {@code [NoHttpResponse, ConnectTimeout, ConnectionRequestTimeout, ServerRetryable]} * Use {@link ClientFaultCause#None} to disable retries. * * @param causes - list of causes @@ -1464,7 +1464,8 @@ public CompletableFuture insert(String tableName, } } } - throw new ClientException("Insert request failed after attempts: " + (retries + 1) + " - Duration: " + (System.nanoTime() - startTime), lastException); + LOG.warn("Insert request failed after attempts: " + (retries + 1) + " - Duration: " + (System.nanoTime() - startTime)); + throw lastException; }; return runAsyncOperation(responseSupplier, settings.getAllSettings()); @@ -1586,8 +1587,8 @@ public CompletableFuture query(String sqlQuery, Map requestConfig, LZ4Factory lz4Factory, - IOCallback writeCallback) throws IOException { + IOCallback writeCallback) throws Exception { if (poolControl != null && timeToPoolVent.get() < System.currentTimeMillis()) { timeToPoolVent.set(System.currentTimeMillis() + POOL_VENT_TIMEOUT); poolControl.closeExpired(); @@ -432,14 +425,10 @@ public ClassicHttpResponse executeRequest(Endpoint server, Map r } catch (UnknownHostException e) { LOG.warn("Host '{}' unknown", server.getBaseURL()); - throw new ClientException("Unknown host", e); + throw e; } catch (ConnectException | NoRouteToHostException e) { LOG.warn("Failed to connect to '{}': {}", server.getBaseURL(), e.getMessage()); - throw new ClientException("Failed to connect", e); - } catch (ConnectionRequestTimeoutException | ServerException | NoHttpResponseException | ClientException | SocketTimeoutException e) { throw e; - } catch (Exception e) { - throw new ClientException(e.getMessage(), e); } } @@ -651,6 +640,12 @@ public boolean shouldRetry(Throwable ex, Map requestSettings) { return retryCauses.contains(ClientFaultCause.SocketTimeout); } + // there are some db retryable error codes + if (ex instanceof ServerException || ex.getCause() instanceof ServerException) { + ServerException se = (ServerException) ex; + return se.isRetryable() && retryCauses.contains(ClientFaultCause.ServerRetryable); + } + return false; } @@ -664,11 +659,17 @@ public RuntimeException wrapException(String message, Exception cause) { if (cause instanceof ConnectionRequestTimeoutException || cause instanceof NoHttpResponseException || cause instanceof ConnectTimeoutException || - cause instanceof ConnectException) { + cause instanceof ConnectException || + cause instanceof UnknownHostException || + cause instanceof NoRouteToHostException) { return new ConnectionInitiationException(message, cause); } - return new ClientException(message, cause); + if (cause instanceof SocketTimeoutException || cause instanceof IOException) { + return new DataTransferException(message, cause); + } + // if we can not identify the exception explicitly we catch as our base exception ClickHouseException + return new ClickHouseException(message, cause); } diff --git a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java index 3499f1dc1..f9bc2e08a 100644 --- a/client-v2/src/test/java/com/clickhouse/client/ClientTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/ClientTests.java @@ -285,7 +285,7 @@ public void testWithOldDefaults() { .enableConnectionPool(true) .setConnectionTTL(-1, MILLIS) .retryOnFailures(ClientFaultCause.NoHttpResponse, ClientFaultCause.ConnectTimeout, - ClientFaultCause.ConnectionRequestTimeout) + ClientFaultCause.ConnectionRequestTimeout, ClientFaultCause.ServerRetryable) .setClientNetworkBufferSize(300_000) .setMaxRetries(3) .allowBinaryReaderToReuseBuffers(false) 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 dbc41304f..d451ec645 100644 --- a/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java +++ b/client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java @@ -1,5 +1,6 @@ package com.clickhouse.client; +import com.clickhouse.client.api.ClickHouseException; import com.clickhouse.client.api.Client; import com.clickhouse.client.api.ClientConfigProperties; import com.clickhouse.client.api.ClientException; @@ -280,7 +281,7 @@ public void testInsertAndNoHttpResponseFailure(String body, int maxRetries, Thro try { function.apply(mockServerClient); - } catch (ClientException e) { + } catch (ConnectionInitiationException e) { e.printStackTrace(); if (!shouldFail) { Assert.fail("Unexpected exception", e); @@ -777,7 +778,8 @@ public void testErrorWithSendProgressHeaders() throws Exception { try (QueryResponse resp = client.query("INSERT INTO test_omm_table SELECT randomString(16) FROM numbers(300000000)", settings).get()) { } catch (ServerException e) { - Assert.assertEquals(e.getCode(), 241); + // 241 - MEMORY_LIMIT_EXCEEDED or 243 -NOT_ENOUGH_SPACE + Assert.assertTrue(e.getCode() == 241 || e.getCode() == 243); } } } 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 b37bfe4a2..903abdf93 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 @@ -4,10 +4,7 @@ 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.ClientConfigProperties; -import com.clickhouse.client.api.ClientException; -import com.clickhouse.client.api.DataTypeUtils; +import com.clickhouse.client.api.*; import com.clickhouse.client.api.command.CommandResponse; import com.clickhouse.client.api.command.CommandSettings; import com.clickhouse.client.api.data_formats.RowBinaryFormatWriter; @@ -26,16 +23,10 @@ import com.clickhouse.client.api.query.QuerySettings; import com.clickhouse.data.ClickHouseFormat; import com.clickhouse.data.ClickHouseVersion; -import com.clickhouse.data.format.BinaryStreamUtils; -import lombok.Data; -import net.jpountz.lz4.LZ4Compressor; -import net.jpountz.lz4.LZ4Factory; -import net.jpountz.lz4.LZ4SafeDecompressor; import org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorOutputStream; import org.apache.commons.compress.compressors.snappy.SnappyCompressorOutputStream; import org.apache.commons.lang3.StringEscapeUtils; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; -import org.testcontainers.shaded.org.checkerframework.checker.units.qual.A; import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -59,7 +50,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.TimeZone; import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.zip.GZIPOutputStream; @@ -242,7 +232,7 @@ public void testInsertingPOJOWithNullValueForNonNullableColumn() throws Exceptio try (InsertResponse response = client.insert(tableName, Collections.singletonList(pojo), settings).get(30, TimeUnit.SECONDS)) { fail("Should have thrown an exception"); - } catch (ClientException e) { + } catch (ClickHouseException e) { e.printStackTrace(); assertTrue(e.getCause() instanceof IllegalArgumentException); }