Update ClickHouseTestHelpers.java

Author: Paultagoras

src/test/java/com/clickhouse/kafka/connect/sink/helper/ClickHouseTestHelpers.java

@@ -61,14 +61,38 @@ public static void query(ClickHouseHelperClient chc, String query) {
             chc.queryV1(query);
         }
     }
-    public static void dropTable(ClickHouseHelperClient chc, String tableName) {
+
+    public static OperationMetrics dropTable(ClickHouseHelperClient chc, String tableName) {
+        for (int i = 0; i < 5; i++) {
+            try {
+                OperationMetrics operationMetrics = dropTableLoop(chc, tableName);
+                if (operationMetrics != null) {
+                    return operationMetrics;
+                }
+            } catch (Exception e) {
+                LOGGER.error("Error while sleeping", e);
+            }
+
+            try {
+                Thread.sleep(30000);//Sleep for 30 seconds
+            } catch (InterruptedException e) {
+                LOGGER.error("Error while sleeping", e);
+            }
+        }
+
+        return null;
+    }
+    private static OperationMetrics dropTableLoop(ClickHouseHelperClient chc, String tableName) {
         String dropTable = String.format("DROP TABLE IF EXISTS `%s`", tableName);
         try {
-            chc.getClient().queryRecords(dropTable).get(CLOUD_TIMEOUT_VALUE, CLOUD_TIMEOUT_UNIT);
+            return chc.getClient().queryRecords(dropTable).get(CLOUD_TIMEOUT_VALUE, CLOUD_TIMEOUT_UNIT).getMetrics();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
+
+
+
     public static OperationMetrics createTable(ClickHouseHelperClient chc, String tableName, String createTableQuery) {
         LOGGER.info("Creating table: {}, Query: {}", tableName, createTableQuery);
         OperationMetrics operationMetrics = createTable(chc, tableName, createTableQuery, new HashMap<>());
@@ -83,18 +107,38 @@ public static OperationMetrics createTable(ClickHouseHelperClient chc, String ta
     }
 
     public static OperationMetrics createTable(ClickHouseHelperClient chc, String tableName, String createTableQuery, Map<String, Serializable> clientSettings) {
+        for (int i = 0; i < 5; i++) {
+            try {
+                OperationMetrics operationMetrics = createTableLoop(chc, tableName, createTableQuery, clientSettings);
+                if (operationMetrics != null) {
+                    return operationMetrics;
+                }
+            } catch (Exception e) {
+                LOGGER.error("Error while sleeping", e);
+            }
+
+            try {
+                Thread.sleep(30000);//Sleep for 30 seconds
+            } catch (InterruptedException e) {
+                LOGGER.error("Error while sleeping", e);
+            }
+        }
+
+        return null;
+    }
+    private static OperationMetrics createTableLoop(ClickHouseHelperClient chc, String tableName, String createTableQuery, Map<String, Serializable> clientSettings) {
         final String createTableQueryTmp = String.format(createTableQuery, tableName);
         QuerySettings settings = new QuerySettings();
         for (Map.Entry<String, Serializable> entry : clientSettings.entrySet()) {
             settings.setOption(entry.getKey(), entry.getValue());
         }
         try {
-            Records records = chc.getClient().queryRecords(createTableQueryTmp, settings).get(CLOUD_TIMEOUT_VALUE, CLOUD_TIMEOUT_UNIT);
-            return records.getMetrics();
+            return chc.getClient().queryRecords(createTableQueryTmp, settings).get(CLOUD_TIMEOUT_VALUE, CLOUD_TIMEOUT_UNIT).getMetrics();
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
+
     public static List<JSONObject> getAllRowsAsJson(ClickHouseHelperClient chc, String tableName)  {
         String query = String.format("SELECT * FROM `%s`", tableName);
         QuerySettings querySettings = new QuerySettings();