Added logic to commit batch in single threaded mode

Altinity · Sep 9, 2024 · a603892 · a603892
1 parent ab888d7
commit a603892
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/.../src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java b/.../src/main/java/com/altinity/clickhouse/sink/connector/executor/ClickHouseBatchWriter.java
@@ -161,6 +161,14 @@ public void persistRecords(List<ClickHouseStruct> records) {
                         //throw new RuntimeException(e);
                         log.error("Error marking records as processed"+ e);
                     }
+
+                    if(record.isLastRecordInBatch()) {
+                        try {
+                            record.getCommitter().markBatchFinished();
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException(e);
+                        }
+                    }
                 });
             }
         } catch(Exception e) {