commit 888ac76712f942b3ac4b1b8b03357d779f742d3a
author GoGoWen <82132356+GoGoWen@users.noreply.github.com> Wed May 10 15:02:02 2023 +0800
committer GitHub <noreply@github.com> Wed May 10 15:02:02 2023 +0800
tree 1932ea8e1d4e830e9683ca31174a3413eec7e5b3
parent acf6e4582d8e651eff6174d05aa56f7681434645

fix interrupt issue (#140)

flink-doris-connector/src/main/java/org/apache/doris/flink/sink/writer/DorisWriter.java

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
index 8ba2050..711f765 100644
--- 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
@@ -182,19 +182,25 @@
                 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();
-            }
 
-            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();
+            // 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();
+            }
         }
     }