[bugfix][cdc-base] Fix cdc base shutdown thread not cleared

apache · Mar 10, 2023 · daa9b4f · daa9b4f
1 parent add75d7
commit daa9b4f
Show file tree

Hide file tree

Showing 7 changed files with 30 additions and 1 deletion.
diff --git a/.../main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/FetchTask.java b/.../main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/FetchTask.java
@@ -41,6 +41,9 @@ public interface FetchTask<Split> {
     /** Returns current task is running or not. */
     boolean isRunning();
 
+    /** Close this task */
+    void shutdown();
+
     /** Returns the split that the task used. */
     Split getSplit();
 

diff --git a/...he/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java b/...he/seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceScanFetcher.java
@@ -192,6 +192,9 @@ private void checkReadException() {
     @Override
     public void close() {
         try {
+            if (snapshotSplitReadTask != null) {
+                snapshotSplitReadTask.shutdown();
+            }
             if (executorService != null) {
                 executorService.shutdown();
                 if (executorService.awaitTermination(

diff --git a/.../seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java b/.../seatunnel/connectors/cdc/base/source/reader/external/IncrementalSourceStreamFetcher.java
@@ -121,9 +121,12 @@ private void checkReadException() {
     @Override
     public void close() {
         try {
+            if (streamFetchTask != null) {
+                streamFetchTask.shutdown();
+            }
             if (executorService != null) {
                 executorService.shutdown();
-                if (executorService.awaitTermination(
+                if (!executorService.awaitTermination(
                         READER_CLOSE_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
                     log.warn(
                             "Failed to close the stream fetcher in {} seconds.",

diff --git a/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/binlog/MySqlBinlogFetchTask.java b/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/binlog/MySqlBinlogFetchTask.java
@@ -81,6 +81,11 @@ public boolean isRunning() {
         return taskRunning;
     }
 
+    @Override
+    public void shutdown() {
+        taskRunning = false;
+    }
+
     @Override
     public SourceSplitBase getSplit() {
         return split;

diff --git a/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/scan/MySqlSnapshotFetchTask.java b/...unnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/scan/MySqlSnapshotFetchTask.java
@@ -155,6 +155,11 @@ public boolean isRunning() {
         return taskRunning;
     }
 
+    @Override
+    public void shutdown() {
+        taskRunning = false;
+    }
+
     @Override
     public SourceSplitBase getSplit() {
         return split;

diff --git a/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/scan/SqlServerSnapshotFetchTask.java b/...s/seatunnel/cdc/sqlserver/source/source/reader/fetch/scan/SqlServerSnapshotFetchTask.java
@@ -166,6 +166,11 @@ public boolean isRunning() {
         return taskRunning;
     }
 
+    @Override
+    public void shutdown() {
+        taskRunning = false;
+    }
+
     @Override
     public SourceSplitBase getSplit() {
         return split;

diff --git a/...sqlserver/source/source/reader/fetch/transactionlog/SqlServerTransactionLogFetchTask.java b/...sqlserver/source/source/reader/fetch/transactionlog/SqlServerTransactionLogFetchTask.java
@@ -78,6 +78,11 @@ public boolean isRunning() {
         return taskRunning;
     }
 
+    @Override
+    public void shutdown() {
+        taskRunning = false;
+    }
+
     @Override
     public SourceSplitBase getSplit() {
         return split;