fix block

JNSimba · Jan 9, 2025 · 2cc2c36 · 2cc2c36
1 parent df088ba
commit 2cc2c36
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 2 deletions.
diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/HttpUtil.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/HttpUtil.java
@@ -60,6 +60,7 @@ protected boolean isRedirectable(String method) {
                                         return true;
                                     }
                                 })
+                        .setRetryHandler((exception, executionCount, context) -> false)
                         .setConnectionReuseStrategy(NoConnectionReuseStrategy.INSTANCE)
                         .setDefaultRequestConfig(
                                 RequestConfig.custom()

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisStreamLoad.java
@@ -115,14 +115,18 @@ public DorisStreamLoad(
         this.streamLoadProp = executionOptions.getStreamLoadProp();
         this.enableDelete = executionOptions.getDeletable();
         this.httpClient = httpClient;
+        String threadName =
+                String.format(
+                        "stream-load-upload-%s-%s",
+                        labelGenerator.getSubtaskId(), labelGenerator.getTableIdentifier());
         this.executorService =
                 new ThreadPoolExecutor(
                         1,
                         1,
                         0L,
                         TimeUnit.MILLISECONDS,
                         new LinkedBlockingQueue<>(),
-                        new ExecutorThreadFactory("stream-load-upload"));
+                        new ExecutorThreadFactory(threadName));
         this.recordStream =
                 new RecordStream(
                         executionOptions.getBufferSize(),
@@ -347,11 +351,20 @@ public void startLoad(String label, boolean isResume) throws IOException {
             } else {
                 executeMessage = "table " + table + " start execute load for label " + label;
             }
+            Thread currentThread = Thread.currentThread();
             pendingLoadFuture =
                     executorService.submit(
                             () -> {
                                 LOG.info(executeMessage);
-                                return httpClient.execute(putBuilder.build());
+                                try {
+                                    return httpClient.execute(putBuilder.build());
+                                } catch (Exception e) {
+                                    LOG.error("Failed to execute load, cause ", e);
+                                    // When an HTTP error occurs, the main thread should be
+                                    // interrupted to prevent blocking
+                                    currentThread.interrupt();
+                                    throw e;
+                                }
                             });
         } catch (Exception e) {
             String err;

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/LabelGenerator.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/LabelGenerator.java
@@ -91,4 +91,12 @@ public String getConcatLabelPrefix() {
         String concatPrefix = String.format("%s_%s_%s", labelPrefix, tableIdentifier, subtaskId);
         return concatPrefix;
     }
+
+    public int getSubtaskId() {
+        return subtaskId;
+    }
+
+    public String getTableIdentifier() {
+        return tableIdentifier;
+    }
 }