delay init scheduled

Author: todd5167

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -77,7 +77,6 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private transient ScheduledThreadPoolExecutor timerService;
 
     public RetractJDBCOutputFormat() {
-        this.timerService = new ScheduledThreadPoolExecutor(1);
     }
 
     @Override
@@ -96,6 +95,8 @@ public void open(int taskNumber, int numTasks) throws IOException {
         try {
             establishConnection();
             initMetric();
+
+            this.timerService = new ScheduledThreadPoolExecutor(1);
             if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
                 if (isReplaceInsertQuery()) {
                     insertQuery = dbSink.buildUpdateSql(tableName, Arrays.asList(dbSink.getFieldNames()), realIndexes, fullField);
@@ -174,7 +175,7 @@ private void insertWrite(Row row) throws SQLException {
             upload.executeBatch();
             batchCount.set(0);
 
-            if (scheduledFuture != null) {
+            if (scheduledFuture != null && !scheduledFuture.isCancelled()) {
                 scheduledFuture.cancel(true);
             }
         }