pick pr 140 102

JNSimba · Apr 26, 2024 · a1d3525 · a1d3525
1 parent 5dd0774
commit a1d3525
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 13 deletions.
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
@@ -78,7 +78,7 @@ public class DorisStreamLoad implements Serializable {
     private final boolean enableDelete;
     private final Properties streamLoadProp;
     private final RecordStream recordStream;
-    private Future<CloseableHttpResponse> pendingLoadFuture;
+    private volatile Future<CloseableHttpResponse> pendingLoadFuture;
     private final CloseableHttpClient httpClient;
     private final ExecutorService executorService;
     private boolean loadBatchFirstRecord;

diff --git a/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java b/flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java
@@ -167,19 +167,24 @@ private void checkDone() {
                 LOG.debug("not loading, skip timer checker");
                 return;
             }
-            // TODO: introduce cache for reload instead of throwing exceptions.
-            String errorMsg;
-            try {
-                RespContent content = dorisStreamLoad.handlePreCommitResponse(dorisStreamLoad.getPendingLoadFuture().get());
-                errorMsg = content.getMessage();
-            } catch (Exception e) {
-                errorMsg = e.getMessage();
+            // double check to interrupt when loading is true and dorisStreamLoad.getPendingLoadFuture().isDone
+            // fix issue #139
+            if (dorisStreamLoad.getPendingLoadFuture() != null
+                    && dorisStreamLoad.getPendingLoadFuture().isDone()) {
+                // TODO: introduce cache for reload instead of throwing exceptions.
+                String errorMsg;
+                try {
+                    RespContent content = dorisStreamLoad.handlePreCommitResponse(dorisStreamLoad.getPendingLoadFuture().get());
+                    errorMsg = content.getMessage();
+                } catch (Exception e) {
+                    errorMsg = e.getMessage();
+                }
+
+                loadException = new StreamLoadException(errorMsg);
+                LOG.error("stream load finished unexpectedly, interrupt worker thread! {}", errorMsg);
+                // set the executor thread interrupted in case blocking in write data.
+                executorThread.interrupt();
             }
-
-            loadException = new StreamLoadException(errorMsg);
-            LOG.error("stream load finished unexpectedly, interrupt worker thread! {}", errorMsg);
-            // set the executor thread interrupted in case blocking in write data.
-            executorThread.interrupt();
         }
     }