Merge remote-tracking branch 'origin/1.8_release_3.10.x' into tmp_1.8_4.0_x_merge

# Conflicts:
#	flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java

Author: yanghuaiGit

flinkx-rdb/flinkx-rdb-reader/src/main/java/com.dtstack.flinkx.rdb.inputformat/JdbcInputFormat.java

@@ -57,6 +57,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -173,7 +174,9 @@ public void openInternal(InputSplit inputSplit) throws IOException {
         querySql = buildQuerySql(inputSplit);
         try {
             executeQuery(((JdbcInputSplit) inputSplit).getStartLocation());
-            columnCount = resultSet.getMetaData().getColumnCount();
+           if(!resultSet.isClosed()){
+               columnCount = resultSet.getMetaData().getColumnCount();
+           }
         } catch (SQLException se) {
             throw new IllegalArgumentException("open() failed." + se.getMessage(), se);
         }
@@ -213,11 +216,11 @@ public boolean reachedEnd() {
             if (incrementConfig.isPolling()) {
                 try {
                     TimeUnit.MILLISECONDS.sleep(incrementConfig.getPollingInterval());
-                    //间隔轮询检测数据库连接是否断开，超时时间三秒，断开后自动重连
-                    if(!dbConn.isValid(3)){
+                    //间隔轮询检测数据库连接是否断开，断开重新连接
+                    if(Objects.isNull(dbConn) || dbConn.isClosed()){
                         dbConn = DbUtil.getConnection(dbUrl, username, password);
                         //重新连接后还是不可用则认为数据库异常，任务失败
-                        if(!dbConn.isValid(3)){
+                        if(Objects.isNull(dbConn) || dbConn.isClosed()){
                             String message = String.format("cannot connect to %s, username = %s, please check %s is available.", dbUrl, username, databaseInterface.getDatabaseType());
                             LOG.error(message);
                             throw new RuntimeException(message);
@@ -246,6 +249,7 @@ public boolean reachedEnd() {
     @Override
     public Row nextRecordInternal(Row row) throws IOException {
         try {
+            updateColumnCount();
             if (!ConstantValue.STAR_SYMBOL.equals(metaColumns.get(0).getName())) {
                 for (int i = 0; i < columnCount; i++) {
                     Object val = row.getField(i);
@@ -888,4 +892,19 @@ protected List<String> analyzeTable(String dbUrl, String username, String passwo
                                         String table, List<MetaColumn> metaColumns) {
         return DbUtil.analyzeTable(dbUrl, username, password, databaseInterface, table, metaColumns);
     }
+
+    /**
+     * 兼容db2 在间隔轮训场景 且第一次读取时没有任何数据
+     * 在openInternal方法调用时 由于数据库没有数据，db2会自动关闭resultSet，因此只有在间隔轮训中某次读取到数据之后，进行更新columnCount
+     * @throws SQLException
+     */
+    private  void updateColumnCount() throws SQLException {
+        if(columnCount == 0){
+            columnCount =resultSet.getMetaData().getColumnCount();
+            boolean splitWithRowCol = numPartitions > 1 && StringUtils.isNotEmpty(splitKey) && splitKey.contains("(");
+            if (splitWithRowCol) {
+                columnCount = columnCount - 1;
+            }
+        }
+    }
 }