update

JNSimba · Nov 15, 2024 · 9f7d98a · 9f7d98a
1 parent fddfa30
commit 9f7d98a
Showing 1 changed file with 6 additions and 1 deletion.
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
@@ -254,6 +254,10 @@ private synchronized boolean doFlush(
     }
 
     private synchronized boolean flush(String bufferKey, boolean waitUtilDone) {
+        if (bufferMap.isEmpty()) {
+            // bufferMap may have been flushed by other threads
+            return false;
+        }
         if (null == bufferKey) {
             boolean flush = false;
             for (String key : bufferMap.keySet()) {
@@ -270,7 +274,7 @@ private synchronized boolean flush(String bufferKey, boolean waitUtilDone) {
         } else if (bufferMap.containsKey(bufferKey)) {
             flushBuffer(bufferKey);
         } else {
-            throw new DorisBatchLoadException("buffer not found for key: " + bufferKey);
+            LOG.warn("buffer not found for key: {}, may be already flushed.", bufferKey);
         }
         if (waitUtilDone) {
             waitAsyncLoadFinish();
@@ -281,6 +285,7 @@ private synchronized boolean flush(String bufferKey, boolean waitUtilDone) {
     private synchronized void flushBuffer(String bufferKey) {
         BatchRecordBuffer buffer = bufferMap.get(bufferKey);
         buffer.setLabelName(labelGenerator.generateBatchLabel(buffer.getTable()));
+        LOG.debug("flush buffer for key {} with label {}", bufferKey, buffer.getLabelName());
         putRecordToFlushQueue(buffer);
         bufferMap.remove(bufferKey);
     }