Merge branch 'feat_1.8_impalaPartition' into 1.8_test_3.10.x

Author: dapeng

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;
     }
 }

oracle/oracle-side/oracle-async-side/src/main/java/com/dtstack/flink/sql/side/oracle/OracleAsyncReqRow.java

@@ -56,7 +56,7 @@ public void open(Configuration parameters) throws Exception {
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
                 .put("provider_class", DT_PROVIDER_CLASS)
-                .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
+                .put("preferred_test_query", "select 1 from dual")
                 .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
                 .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
 

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java

@@ -78,7 +78,7 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     public final static String DT_PROVIDER_CLASS = "com.dtstack.flink.sql.side.rdb.provider.DTC3P0DataSourceProvider";
 
-    public final static String PREFERRED_TEST_QUERY_SQL = "select 1 from dual";
+    public final static String PREFERRED_TEST_QUERY_SQL = "select 1";
 
     private transient SQLClient rdbSqlClient;
 

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
          */