[Fix](batch) Fixed an issue where the writer might be blocked in batc…

…h mode (apache#506)
JNSimba · Nov 6, 2024 · 57003ef · 57003ef
1 parent 6f3efe0
commit 57003ef
Showing 1 changed file with 17 additions and 5 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
@@ -198,6 +198,7 @@ public synchronized void writeRecord(String database, String table, byte[] recor
             lock.lock();
             try {
                 while (currentCacheBytes.get() >= maxBlockedBytes) {
+                    checkFlushException();
                     LOG.info(
                             "Cache full, waiting for flush, currentBytes: {}, maxBlockedBytes: {}",
                             currentCacheBytes.get(),
@@ -486,11 +487,22 @@ public void load(String label, BatchRecordBuffer buffer) throws IOException {
                                 putBuilder.setLabel(label + "_" + retry);
                                 reason = respContent.getMessage();
                             } else {
-                                String errMsg =
-                                        String.format(
-                                                "stream load error: %s, see more in %s",
-                                                respContent.getMessage(),
-                                                respContent.getErrorURL());
+                                String errMsg = null;
+                                if (StringUtils.isBlank(respContent.getMessage())
+                                        && StringUtils.isBlank(respContent.getErrorURL())) {
+                                    // sometimes stream load will not return message
+                                    errMsg =
+                                            String.format(
+                                                    "stream load error, response is %s",
+                                                    loadResult);
+                                    throw new DorisBatchLoadException(errMsg);
+                                } else {
+                                    errMsg =
+                                            String.format(
+                                                    "stream load error: %s, see more in %s",
+                                                    respContent.getMessage(),
+                                                    respContent.getErrorURL());
+                                }
                                 throw new DorisBatchLoadException(errMsg);
                             }
                         }