[Improve] Improve doris stream load client side error message (apache#6688)

Author: Hisoka-X

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java

@@ -58,15 +58,13 @@ public class DorisSinkWriter
             new ArrayList<>(Arrays.asList(LoadStatus.SUCCESS, LoadStatus.PUBLISH_TIMEOUT));
     private long lastCheckpointId;
     private DorisStreamLoad dorisStreamLoad;
-    volatile boolean loading;
     private final DorisConfig dorisConfig;
     private final String labelPrefix;
     private final LabelGenerator labelGenerator;
     private final int intervalTime;
     private final DorisSerializer serializer;
     private final CatalogTable catalogTable;
     private final ScheduledExecutorService scheduledExecutorService;
-    private Thread executorThread;
     private volatile Exception loadException = null;
 
     public DorisSinkWriter(
@@ -94,7 +92,6 @@ public DorisSinkWriter(
                         1, new ThreadFactoryBuilder().setNameFormat("stream-load-check").build());
         this.serializer = createSerializer(dorisConfig, catalogTable.getSeaTunnelRowType());
         this.intervalTime = dorisConfig.getCheckInterval();
-        this.loading = false;
         this.initializeLoad();
     }
 
@@ -123,7 +120,7 @@ private void initializeLoad() {
 
     @Override
     public void write(SeaTunnelRow element) throws IOException {
-        checkLoadExceptionAndResetThread();
+        checkLoadException();
         byte[] serialize =
                 serializer.serialize(
                         dorisConfig.isNeedsUnsupportedTypeCasting()
@@ -154,7 +151,6 @@ public Optional<DorisCommitInfo> prepareCommit() throws IOException {
 
     private RespContent flush() throws IOException {
         // disable exception checker before stop load.
-        loading = false;
         checkState(dorisStreamLoad != null);
         RespContent respContent = dorisStreamLoad.stopLoad();
         if (respContent != null && !DORIS_SUCCESS_STATUS.contains(respContent.getStatus())) {
@@ -177,7 +173,6 @@ public List<DorisSinkState> snapshotState(long checkpointId) {
 
     private void startLoad(String label) {
         this.dorisStreamLoad.startLoad(label);
-        this.loading = true;
     }
 
     @Override
@@ -194,37 +189,19 @@ public void abortPrepare() {
     private void checkDone() {
         // the load future is done and checked in prepareCommit().
         // this will check error while loading.
+        String errorMsg;
         log.debug("start timer checker, interval {} ms", intervalTime);
-        if (dorisStreamLoad.getPendingLoadFuture() != null
-                && dorisStreamLoad.getPendingLoadFuture().isDone()) {
-            if (!loading) {
-                log.debug("not loading, skip timer checker");
-                return;
-            }
-            String errorMsg;
-            try {
-                RespContent content =
-                        dorisStreamLoad.handlePreCommitResponse(
-                                dorisStreamLoad.getPendingLoadFuture().get());
-                errorMsg = content.getMessage();
-            } catch (Exception e) {
-                errorMsg = e.getMessage();
-            }
-
+        if ((errorMsg = dorisStreamLoad.getLoadFailedMsg()) != null) {
+            log.error("stream load finished unexpectedly: {}", errorMsg);
             loadException =
                     new DorisConnectorException(
                             DorisConnectorErrorCode.STREAM_LOAD_FAILED, errorMsg);
-            log.error("stream load finished unexpectedly, interrupt worker thread! {}", errorMsg);
-            // set the executor thread interrupted in case blocking in write data.
-            executorThread.interrupt();
         }
     }
 
-    private void checkLoadExceptionAndResetThread() {
+    private void checkLoadException() {
         if (loadException != null) {
             throw new RuntimeException("error while loading data.", loadException);
-        } else {
-            executorThread = Thread.currentThread();
         }
     }
 

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisStreamLoad.java

@@ -78,6 +78,7 @@ public class DorisStreamLoad implements Serializable {
     private final CloseableHttpClient httpClient;
     private final ExecutorService executorService;
     private volatile boolean loadBatchFirstRecord;
+    private volatile boolean loading = false;
     private String label;
     private long recordCount = 0;
 
@@ -199,7 +200,25 @@ public long getRecordCount() {
         return recordCount;
     }
 
-    public RespContent handlePreCommitResponse(CloseableHttpResponse response) throws Exception {
+    public String getLoadFailedMsg() {
+        if (!loading) {
+            return null;
+        }
+        if (this.getPendingLoadFuture() != null && this.getPendingLoadFuture().isDone()) {
+            String errorMessage;
+            try {
+                errorMessage = handlePreCommitResponse(pendingLoadFuture.get()).getMessage();
+            } catch (Exception e) {
+                errorMessage = e.getMessage();
+            }
+            recordStream.setErrorMessageByStreamLoad(errorMessage);
+            return errorMessage;
+        } else {
+            return null;
+        }
+    }
+
+    private RespContent handlePreCommitResponse(CloseableHttpResponse response) throws Exception {
         final int statusCode = response.getStatusLine().getStatusCode();
         if (statusCode == HTTP_TEMPORARY_REDIRECT && response.getEntity() != null) {
             String loadResult = EntityUtils.toString(response.getEntity());
@@ -211,6 +230,7 @@ public RespContent handlePreCommitResponse(CloseableHttpResponse response) throw
     }
 
     public RespContent stopLoad() throws IOException {
+        loading = false;
         if (pendingLoadFuture != null) {
             log.info("stream load stopped.");
             recordStream.endInput();
@@ -230,6 +250,7 @@ public void startLoad(String label) {
         loadBatchFirstRecord = true;
         recordCount = 0;
         this.label = label;
+        this.loading = true;
     }
 
     private void startStreamLoad() {

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/RecordBuffer.java

@@ -17,6 +17,10 @@
 
 package org.apache.seatunnel.connectors.doris.sink.writer;
 
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorErrorCode;
+import org.apache.seatunnel.connectors.doris.exception.DorisConnectorException;
+
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
@@ -25,18 +29,21 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
 
 import static com.google.common.base.Preconditions.checkState;
 
 /** Channel of record stream and HTTP data stream. */
 @Slf4j
 public class RecordBuffer {
-    BlockingQueue<ByteBuffer> writeQueue;
-    BlockingQueue<ByteBuffer> readQueue;
-    int bufferCapacity;
-    int queueSize;
-    ByteBuffer currentWriteBuffer;
-    ByteBuffer currentReadBuffer;
+    private final BlockingQueue<ByteBuffer> writeQueue;
+    private final BlockingQueue<ByteBuffer> readQueue;
+    private final int bufferCapacity;
+    private final int queueSize;
+    private ByteBuffer currentWriteBuffer;
+    private ByteBuffer currentReadBuffer;
+    // used to check stream load error by stream load thread
+    @Setter private volatile String errorMessageByStreamLoad;
 
     public RecordBuffer(int capacity, int queueSize) {
         log.info("init RecordBuffer capacity {}, count {}", capacity, queueSize);
@@ -76,7 +83,11 @@ public void stopBufferData() throws IOException {
                 currentWriteBuffer = null;
             }
             if (!isEmpty) {
-                ByteBuffer byteBuffer = writeQueue.take();
+                ByteBuffer byteBuffer = null;
+                while (byteBuffer == null) {
+                    checkErrorMessageByStreamLoad();
+                    byteBuffer = writeQueue.poll(100, TimeUnit.MILLISECONDS);
+                }
                 ((Buffer) byteBuffer).flip();
                 checkState(byteBuffer.limit() == 0);
                 readQueue.put(byteBuffer);
@@ -89,8 +100,9 @@ public void stopBufferData() throws IOException {
     public void write(byte[] buf) throws InterruptedException {
         int wPos = 0;
         do {
-            if (currentWriteBuffer == null) {
-                currentWriteBuffer = writeQueue.take();
+            while (currentWriteBuffer == null) {
+                checkErrorMessageByStreamLoad();
+                currentWriteBuffer = writeQueue.poll(100, TimeUnit.MILLISECONDS);
             }
             int available = currentWriteBuffer.remaining();
             int nWrite = Math.min(available, buf.length - wPos);
@@ -105,14 +117,15 @@ public void write(byte[] buf) throws InterruptedException {
     }
 
     public int read(byte[] buf) throws InterruptedException {
-        if (currentReadBuffer == null) {
-            currentReadBuffer = readQueue.take();
+        while (currentReadBuffer == null) {
+            checkErrorMessageByStreamLoad();
+            currentReadBuffer = readQueue.poll(100, TimeUnit.MILLISECONDS);
         }
         // add empty buffer as end flag
         if (currentReadBuffer.limit() == 0) {
             recycleBuffer(currentReadBuffer);
             currentReadBuffer = null;
-            checkState(readQueue.size() == 0);
+            checkState(readQueue.isEmpty());
             return -1;
         }
         int available = currentReadBuffer.remaining();
@@ -125,16 +138,17 @@ public int read(byte[] buf) throws InterruptedException {
         return nRead;
     }
 
-    private void recycleBuffer(ByteBuffer buffer) throws InterruptedException {
-        ((Buffer) buffer).clear();
-        writeQueue.put(buffer);
-    }
-
-    public int getWriteQueueSize() {
-        return writeQueue.size();
+    private void checkErrorMessageByStreamLoad() {
+        if (errorMessageByStreamLoad != null) {
+            throw new DorisConnectorException(
+                    DorisConnectorErrorCode.STREAM_LOAD_FAILED, errorMessageByStreamLoad);
+        }
     }
 
-    public int getReadQueueSize() {
-        return readQueue.size();
+    private void recycleBuffer(ByteBuffer buffer) throws InterruptedException {
+        ((Buffer) buffer).clear();
+        while (!writeQueue.offer(buffer, 100, TimeUnit.MILLISECONDS)) {
+            checkErrorMessageByStreamLoad();
+        }
     }
 }

Diff for: seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/RecordStream.java

@@ -57,4 +57,8 @@ public void write(byte[] buff) throws IOException {
             throw new RuntimeException(e);
         }
     }
+
+    public void setErrorMessageByStreamLoad(String errorMessageByStreamLoad) {
+        recordBuffer.setErrorMessageByStreamLoad(errorMessageByStreamLoad);
+    }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisErrorIT.java

@@ -79,6 +79,17 @@ public void testDoris(TestContainer container) throws InterruptedException, Exec
         Thread.sleep(10 * 1000);
         super.container.stop();
         Assertions.assertNotEquals(0, future.get().getExitCode());
+        Assertions.assertTrue(
+                future.get()
+                        .getStderr()
+                        .contains(
+                                "Caused by: org.apache.seatunnel.connectors.doris.exception.DorisConnectorException: ErrorCode:[Doris-01], ErrorDescription:[stream load error]"));
+        Assertions.assertTrue(
+                future.get()
+                        .getStderr()
+                        .contains(
+                                "at org.apache.seatunnel.connectors.doris.sink.writer.RecordBuffer.checkErrorMessageByStreamLoad"));
+        log.info("doris error log: \n" + future.get().getStderr());
         super.container.start();
         // wait for the container to restart
         given().ignoreExceptions()

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_source_and_doris_sink_timeout_error.conf

@@ -18,13 +18,13 @@
 env{
   parallelism = 1
   job.mode = "BATCH"
+  job.retry.times = 0
 }
 
 source{
   FakeSource {
-    row.num = 100
+    row.num = 1000
     split.num = 10
-    split.read-interval = 10000
     string.length = 1
     schema = {
       fields {
@@ -58,6 +58,10 @@ sink{
           password = ""
           table.identifier = "e2e_sink.doris_e2e_table"
           sink.enable-2pc = "true"
+          // stuck in get RecordBuffer
+          sink.buffer-size = 2
+          sink.buffer-count = 2
+
           sink.label-prefix = "test_json"
           doris.config = {
               format="json"