[Bugfix][Clickhouse] Fix clickhouse sink flush bug (#5448)

* [Bug][connector-cdc-mysql] mysql connections and memory of jvm increased abnormally (#5008) * [bugfix][connector-cdc-mysql] reset the listener of binaryLogClient before fetch task start (#5008) * [Bugfix][Clickhouse] fix when the checkpoint triggers flush, the connection is closed, causing subsequent data writing to fail --------- Co-authored-by: dengjunjie <[email protected]>
apache · Sep 12, 2023 · cef03f6 · cef03f6
1 parent 38b6d6e
commit cef03f6
Showing 1 changed file with 8 additions and 1 deletion.
diff --git a/...rg/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkWriter.java b/...rg/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/ClickhouseSinkWriter.java
@@ -90,7 +90,14 @@ public void write(SeaTunnelRow element) throws IOException {
 
     @Override
     public Optional<CKCommitInfo> prepareCommit() throws IOException {
-        flush();
+        for (ClickhouseBatchStatement batchStatement : statementMap.values()) {
+            JdbcBatchStatementExecutor statement = batchStatement.getJdbcBatchStatementExecutor();
+            IntHolder intHolder = batchStatement.getIntHolder();
+            if (intHolder.getValue() > 0) {
+                flush(statement);
+                intHolder.setValue(0);
+            }
+        }
         return Optional.empty();
     }