Merge branch 'hotfix_1.8_3.10.x_26972' into 1.8_3.10_zy

Author: dapeng

core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java

@@ -22,6 +22,7 @@
 
 import com.dtstack.flink.sql.enums.ECacheContentType;
 import com.dtstack.flink.sql.enums.ECacheType;
+import com.dtstack.flink.sql.factory.DTThreadFactory;
 import com.dtstack.flink.sql.metric.MetricConstant;
 import com.dtstack.flink.sql.side.cache.AbstractSideCache;
 import com.dtstack.flink.sql.side.cache.CacheObj;
@@ -68,6 +69,7 @@ public abstract class BaseAsyncReqRow extends RichAsyncFunction<CRow, CRow> impl
     private int timeOutNum = 0;
     protected BaseSideInfo sideInfo;
     protected transient Counter parseErrorRecords;
+    private transient ThreadPoolExecutor cancelExecutor;
 
     public BaseAsyncReqRow(BaseSideInfo sideInfo){
         this.sideInfo = sideInfo;
@@ -82,6 +84,8 @@ public void open(Configuration parameters) throws Exception {
         super.open(parameters);
         initCache();
         initMetric();
+        cancelExecutor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100000),
+                new DTThreadFactory("cancel-timer-executor"));
         LOG.info("async dim table config info: {} ", sideInfo.getSideTableInfo().toString());
     }
 
@@ -248,12 +252,11 @@ public void onProcessingTime(long timestamp) throws Exception {
     }
 
     protected void cancelTimerWhenComplete(ResultFuture<CRow> resultFuture, ScheduledFuture<?> timerFuture){
-        ThreadPoolExecutor executors = new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
         if(resultFuture instanceof StreamRecordQueueEntry){
             StreamRecordQueueEntry streamRecordBufferEntry = (StreamRecordQueueEntry) resultFuture;
             streamRecordBufferEntry.onComplete((Object value) -> {
                 timerFuture.cancel(true);
-            },executors);
+            }, cancelExecutor);
         }
     }
 

mongo/mongo-side/mongo-side-core/src/main/java/com/dtstack/flink/sql/side/mongo/table/MongoSideParser.java

@@ -59,7 +59,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         mongoSideTableInfo.setDatabase(MathUtil.getString(props.get(DATABASE_KEY.toLowerCase())));
         mongoSideTableInfo.setUserName(MathUtil.getString(props.get(USER_NAME_KEY.toLowerCase())));
         mongoSideTableInfo.setPassword(MathUtil.getString(props.get(PASSWORD_KEY.toLowerCase())));
-
+        mongoSideTableInfo.check();
         return mongoSideTableInfo;
     }
 }

mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/table/MongoSinkParser.java

@@ -60,7 +60,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         mongoTableInfo.setDatabase(MathUtil.getString(props.get(DATABASE_KEY.toLowerCase())));
         mongoTableInfo.setUserName(MathUtil.getString(props.get(USER_NAME_KEY.toLowerCase())));
         mongoTableInfo.setPassword(MathUtil.getString(props.get(PASSWORD_KEY.toLowerCase())));
-
+        mongoTableInfo.check();
         return mongoTableInfo;
     }
 }

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

@@ -33,7 +33,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -44,6 +46,7 @@
 
 /**
  * An upsert OutputFormat for JDBC.
+ *
  * @author maqi
  */
 public class JDBCUpsertOutputFormat extends AbstractJDBCOutputFormat<Tuple2<Boolean, Row>> {
@@ -103,7 +106,7 @@ public JDBCUpsertOutputFormat(
      *
      * @param taskNumber The number of the parallel instance.
      * @throws IOException Thrown, if the output could not be opened due to an
-     * I/O problem.
+     *                     I/O problem.
      */
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
@@ -167,7 +170,7 @@ public synchronized void writeRecord(Tuple2<Boolean, Row> tuple2) throws IOExcep
 
     private void checkConnectionOpen() {
         try {
-            if (connection.isClosed()) {
+            if (!connection.isValid(10)) {
                 LOG.info("db connection reconnect..");
                 establishConnection();
                 jdbcWriter.prepareStatement(connection);
@@ -270,7 +273,8 @@ public Builder setFieldTypes(int[] fieldTypes) {
         }
 
         /**
-         *   optional, partition Fields
+         * optional, partition Fields
+         *
          * @param partitionFields
          * @return
          */