fix batch block

apache · Nov 13, 2024 · fddfa30 · fddfa30
1 parent 498a7fa
commit fddfa30
Showing 1 changed file with 4 additions and 9 deletions.
diff --git a/...doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java b/...doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
@@ -178,7 +178,7 @@ public DorisBatchStreamLoad(
      * @param record
      * @throws IOException
      */
-    public synchronized void writeRecord(String database, String table, byte[] record) {
+    public void writeRecord(String database, String table, byte[] record) {
         checkFlushException();
         String bufferKey = getTableIdentifier(database, table);
 
@@ -228,15 +228,15 @@ public synchronized void writeRecord(String database, String table, byte[] recor
         }
     }
 
-    public synchronized boolean bufferFullFlush(String bufferKey) {
+    public boolean bufferFullFlush(String bufferKey) {
         return doFlush(bufferKey, false, true);
     }
 
-    public synchronized boolean intervalFlush() {
+    public boolean intervalFlush() {
         return doFlush(null, false, false);
     }
 
-    public synchronized boolean checkpointFlush() {
+    public boolean checkpointFlush() {
         return doFlush(null, true, false);
     }
 
@@ -408,11 +408,6 @@ public void run() {
                             load(bf.getLabelName(), bf);
                         }
                     }
-
-                    if (flushQueue.size() < flushQueueSize) {
-                        // Avoid waiting for 2 rounds of intervalMs
-                        doFlush(null, false, false);
-                    }
                 } catch (Exception e) {
                     LOG.error("worker running error", e);
                     exception.set(e);