[Fix](batch) prevent writer deadlock from currentCacheBytes drift

Author: addu390

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

@@ -202,6 +202,18 @@ public void writeRecord(String database, String table, byte[] record) {
         currentCacheBytes.addAndGet(bytes);
         getLock(bufferKey).readLock().unlock();
 
+        if (flushQueue.size() < executionOptions.getFlushQueueSize()
+                && (buffer.getBufferSizeBytes() >= executionOptions.getBufferFlushMaxBytes()
+                        || buffer.getNumOfRecords() >= executionOptions.getBufferFlushMaxRows())) {
+            boolean flush = bufferFullFlush(bufferKey);
+            LOG.info("trigger flush by buffer full, flush: {}", flush);
+        } else if (buffer.getBufferSizeBytes() >= STREAM_LOAD_MAX_BYTES
+                || buffer.getNumOfRecords() >= STREAM_LOAD_MAX_ROWS) {
+            // The buffer capacity exceeds the stream load limit, flush
+            boolean flush = bufferFullFlush(bufferKey);
+            LOG.info("trigger flush by buffer exceeding the limit, flush: {}", flush);
+        }
+
         if (currentCacheBytes.get() > maxBlockedBytes) {
             lock.lock();
             try {
@@ -220,20 +232,6 @@ public void writeRecord(String database, String table, byte[] record) {
                 lock.unlock();
             }
         }
-
-        // queue has space, flush according to the bufferMaxRows/bufferMaxBytes
-        if (flushQueue.size() < executionOptions.getFlushQueueSize()
-                && (buffer.getBufferSizeBytes() >= executionOptions.getBufferFlushMaxBytes()
-                        || buffer.getNumOfRecords() >= executionOptions.getBufferFlushMaxRows())) {
-            boolean flush = bufferFullFlush(bufferKey);
-            LOG.info("trigger flush by buffer full, flush: {}", flush);
-
-        } else if (buffer.getBufferSizeBytes() >= STREAM_LOAD_MAX_BYTES
-                || buffer.getNumOfRecords() >= STREAM_LOAD_MAX_ROWS) {
-            // The buffer capacity exceeds the stream load limit, flush
-            boolean flush = bufferFullFlush(bufferKey);
-            LOG.info("trigger flush by buffer exceeding the limit, flush: {}", flush);
-        }
     }
 
     public boolean bufferFullFlush(String bufferKey) {
@@ -499,7 +497,7 @@ public void load(String label, BatchRecordBuffer buffer) throws IOException {
                                     OBJECT_MAPPER.readValue(loadResult, RespContent.class);
                             if (DORIS_SUCCESS_STATUS.contains(respContent.getStatus())) {
                                 long cacheByteBeforeFlush =
-                                        currentCacheBytes.getAndAdd(-respContent.getLoadBytes());
+                                        currentCacheBytes.getAndAdd(-buffer.getBufferSizeBytes());
                                 LOG.info(
                                         "load success, cacheBeforeFlushBytes: {}, currentCacheBytes : {}",
                                         cacheByteBeforeFlush,