fix review issue

Author: Matrix42

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -80,6 +80,8 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecord, MySqlSpli
     private final Set<TableId> pureBinlogPhaseTables;
     private Tables.TableFilter capturedTableFilter;
 
+    private static final long EXECUTOR_AWAIT_TERMINATION_TIMEOUT_SECONDS = 5;
+
     public BinlogSplitReader(StatefulTaskContext statefulTaskContext, int subTaskId) {
         this.statefulTaskContext = statefulTaskContext;
         ThreadFactory threadFactory =
@@ -181,7 +183,12 @@ public void close() {
             // while loop in MySqlStreamingChangeEventSource's execute method
             currentTaskRunning = false;
             executor.shutdown();
-            executor.awaitTermination(5, TimeUnit.SECONDS);
+            boolean isShutdown =
+                    executor.awaitTermination(
+                            EXECUTOR_AWAIT_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            if (!isShutdown) {
+                LOG.warn("The thread executor of BinlogSplitReader wasn't shutdown properly.");
+            }
         } catch (Exception e) {
             LOG.error("Close binlog reader error", e);
         }

flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java

@@ -87,6 +87,8 @@ public class SnapshotSplitReader implements DebeziumReader<SourceRecord, MySqlSp
     public AtomicBoolean hasNextElement;
     public AtomicBoolean reachEnd;
 
+    private static final long EXECUTOR_AWAIT_TERMINATION_TIMEOUT_SECONDS = 5;
+
     public SnapshotSplitReader(StatefulTaskContext statefulTaskContext, int subtaskId) {
         this.statefulTaskContext = statefulTaskContext;
         ThreadFactory threadFactory =
@@ -330,7 +332,12 @@ public void close() {
                 statefulTaskContext.getBinaryLogClient().disconnect();
             }
             executor.shutdown();
-            executor.awaitTermination(5, TimeUnit.SECONDS);
+            boolean isShutdown =
+                    executor.awaitTermination(
+                            EXECUTOR_AWAIT_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            if (!isShutdown) {
+                LOG.warn("The thread executor of SnapshotSplitReader wasn't shutdown properly.");
+            }
         } catch (Exception e) {
             LOG.error("Close snapshot reader error", e);
         }

flink-connector-mysql-cdc/src/test/java/com/ververica/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java

@@ -443,6 +443,8 @@ private List<String> readBinlogSplits(
         }
         snapshotSplitReader.close();
 
+        assertExecutorIsTerminated(snapshotSplitReader);
+
         // step-2: create binlog split according the finished snapshot splits
         List<FinishedSnapshotSplitInfo> finishedSplitsInfo =
                 getFinishedSplitsInfo(sqlSplits, snapshotRecords);