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

DorisBatchStreamLoad increments currentCacheBytes by the client-side record bytes on insert, but decrements it by respContent.getLoadBytes() on a successful load. Whenever the BE-reported value is smaller than what the client buffered, either by partial_columns=true, compress_type=gz, etc, each load leaks a few bytes from the counter.

Over time the leak accumulates above maxBlockedBytes, so writeRecord parks on block.await() forever even though bufferMap and flushQueue are empty. The job freezes with no exception, only repeating Cache full, waiting for flush and bufferMap is empty, no need to flush null logs.

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,