commit d72d7a95148eefbc9757e7a0613f83bcfe06a236
author 苏承祥 <scx_white@aliyun.com> Thu Aug 10 17:06:09 2023 +0800
committer GitHub <noreply@github.com> Thu Aug 10 17:06:09 2023 +0800
tree a17012166271743205323b01b6c2a16871af2b85
parent be1c5cee79070c0e360918d4a410cafff13e7625

Fix concurrency issues during flush (#177)

Co-authored-by: geyun <suchengxiang.scx@cainiao.com>

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
index 8f2dcc1..1fe217d 100644
--- a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
+++ b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/batch/DorisBatchStreamLoad.java
@@ -124,7 +124,7 @@
      * @param record
      * @throws IOException
      */
-    public void writeRecord(byte[] record) throws InterruptedException {
+    public synchronized void writeRecord(byte[] record) throws InterruptedException {
         checkFlushException();
         if(buffer == null){
             buffer = takeRecordFromWriteQueue();
@@ -137,7 +137,7 @@
         }
     }
 
-    public void flush(boolean waitUtilDone) throws InterruptedException {
+    public synchronized void flush(boolean waitUtilDone) throws InterruptedException {
         checkFlushException();
         if (buffer == null) {
             LOG.debug("buffer is empty, skip flush.");