Fix exception message & extract retry to const.

Author: mzitnik

flink-connector-clickhouse-1.17/src/main/java/org/apache/flink/connector/clickhouse/sink/ClickHouseAsyncSinkSerializer.java

@@ -44,7 +44,7 @@ protected ClickHousePayload deserializeRequestFromStream(long requestSize, DataI
                 throw new IOException("Unsupported version " + version);
             }
         } else {
-            throw new IOException("request size < 0");
+            throw new IOException("Request size: " + requestSize);
         }
     }
 

flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java

@@ -45,6 +45,7 @@ public class ClickHouseSinkTests extends FlinkClusterTests {
     static final long MAX_RECORD_SIZE_IN_BYTES = 1000;
 
     static final int STREAM_PARALLELISM = 5;
+    static final int NUMBER_OF_RETRIES = 10;
 
     private String createSimplePOJOTableSQL(String database, String tableName, int parts_to_throw_insert) {
         String createTable = createSimplePOJOTableSQL(database, tableName);
@@ -511,7 +512,7 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
         env.setParallelism(STREAM_PARALLELISM);
 
         ClickHouseClientConfig clickHouseClientConfig = new ClickHouseClientConfig(getServerURL(), getUsername(), getPassword(), getDatabase(), tableName);
-        clickHouseClientConfig.setNumberOfRetries(10);
+        clickHouseClientConfig.setNumberOfRetries(NUMBER_OF_RETRIES);
         clickHouseClientConfig.setSupportDefault(simpleTableSchema.hasDefaults());
 
         ElementConverter<SimplePOJO, ClickHousePayload> convertorCovid = new ClickHouseConvertor<>(SimplePOJO.class, simplePOJOConvertor);

flink-connector-clickhouse-2.0.0/src/main/java/org/apache/flink/connector/clickhouse/sink/ClickHouseAsyncSinkSerializer.java

@@ -44,7 +44,7 @@ protected ClickHousePayload deserializeRequestFromStream(long requestSize, DataI
                 throw new IOException("Unsupported version " + version);
             }
         } else {
-            throw new IOException("request size < 0");
+            throw new IOException("Request size: " + requestSize);
         }
     }
 

flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java

@@ -45,6 +45,7 @@ public class ClickHouseSinkTests extends FlinkClusterTests {
     static final long MAX_RECORD_SIZE_IN_BYTES = 1000;
 
     static final int STREAM_PARALLELISM = 5;
+    static final int NUMBER_OF_RETRIES = 10;
 
     private String createSimplePOJOTableSQL(String database, String tableName, int parts_to_throw_insert) {
         String createTable = createSimplePOJOTableSQL(database, tableName);
@@ -508,7 +509,7 @@ void SimplePOJODataTooManyPartsTest() throws Exception {
         env.setParallelism(STREAM_PARALLELISM);
 
         ClickHouseClientConfig clickHouseClientConfig = new ClickHouseClientConfig(getServerURL(), getUsername(), getPassword(), getDatabase(), tableName);
-        clickHouseClientConfig.setNumberOfRetries(10);
+        clickHouseClientConfig.setNumberOfRetries(NUMBER_OF_RETRIES);
         clickHouseClientConfig.setSupportDefault(simpleTableSchema.hasDefaults());
 
         ElementConverter<SimplePOJO, ClickHousePayload> convertorCovid = new ClickHouseConvertor<>(SimplePOJO.class, simplePOJOConvertor);