Merge branch '1.8_release_4.0.x' into temp_1.10_4.0.x_merge

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

Author: lvyanquan

flinkx-phoenix5/flinkx-phoenix5-reader/src/main/java/com/dtstack/flinkx/phoenix5/format/Phoenix5InputFormat.java

@@ -239,6 +239,7 @@ public boolean reachedEnd() throws IOException{
      * 获取数据库连接，用于子类覆盖
      * @return connection
      */
+    @Override
     protected Connection getConnection() throws SQLException {
         Field declaredField = ReflectionUtils.getDeclaredField(getClass().getClassLoader(), "ucp");
         assert declaredField != null;

flinkx-postgresql/flinkx-postgresql-writer/src/main/java/com/dtstack/flinkx/postgresql/format/PostgresqlOutputFormat.java

@@ -20,6 +20,7 @@
 import com.dtstack.flinkx.enums.EWriteMode;
 import com.dtstack.flinkx.exception.WriteRecordException;
 import com.dtstack.flinkx.rdb.outputformat.JdbcOutputFormat;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.types.Row;
 import org.postgresql.copy.CopyManager;
 import org.postgresql.core.BaseConnection;
@@ -46,6 +47,8 @@ public class PostgresqlOutputFormat extends JdbcOutputFormat {
 
     private static final String LINE_DELIMITER = "\n";
 
+    private boolean isCopyMode = false;
+
     /**
      * now just add ext insert mode:copy
      */
@@ -63,7 +66,8 @@ protected PreparedStatement prepareTemplates() throws SQLException {
         }
 
         //check is use copy mode for insert
-        if (EWriteMode.INSERT.name().equalsIgnoreCase(mode) && checkIsCopyMode(insertSqlMode)) {
+        isCopyMode = checkIsCopyMode(insertSqlMode);
+        if (EWriteMode.INSERT.name().equalsIgnoreCase(mode) && isCopyMode) {
             copyManager = new CopyManager((BaseConnection) dbConn);
             copySql = String.format(COPY_SQL_TEMPL, table, String.join(",", column), DEFAULT_FIELD_DELIM, DEFAULT_NULL_DELIM);
             return null;
@@ -72,27 +76,10 @@ protected PreparedStatement prepareTemplates() throws SQLException {
         return super.prepareTemplates();
     }
 
-    @Override
-    protected void openInternal(int taskNumber, int numTasks){
-        super.openInternal(taskNumber, numTasks);
-        try {
-            if (batchInterval > 1) {
-                dbConn.setAutoCommit(false);
-            }
-        } catch (Exception e) {
-            LOG.warn("", e);
-        }
-    }
-
     @Override
     protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
-        if(!checkIsCopyMode(insertSqlMode)){
-            if (batchInterval == 1) {
-                super.writeSingleRecordInternal(row);
-            } else {
-                writeSingleRecordCommit(row);
-            }
-
+        if(!isCopyMode){
+            super.writeSingleRecordInternal(row);
             return;
         }
 
@@ -130,31 +117,10 @@ protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
         }
     }
 
-    private void writeSingleRecordCommit(Row row) throws WriteRecordException {
-        try {
-            super.writeSingleRecordInternal(row);
-            try {
-                dbConn.commit();
-            } catch (Exception e) {
-                // 提交失败直接结束任务
-                throw new RuntimeException(e);
-            }
-        } catch (WriteRecordException e) {
-            try {
-                dbConn.rollback();
-            } catch (Exception e1) {
-                // 回滚失败直接结束任务
-                throw new RuntimeException(e);
-            }
-
-            throw e;
-        }
-    }
-
     @Override
     protected void writeMultipleRecordsInternal() throws Exception {
-        if(!checkIsCopyMode(insertSqlMode)){
-            writeMultipleRecordsCommit();
+        if(!isCopyMode){
+            super.writeMultipleRecordsInternal();
             return;
         }
 
@@ -197,16 +163,6 @@ protected void writeMultipleRecordsInternal() throws Exception {
         }
     }
 
-    private void writeMultipleRecordsCommit() throws Exception {
-        try {
-            super.writeMultipleRecordsInternal();
-            dbConn.commit();
-        } catch (Exception e){
-            dbConn.rollback();
-            throw e;
-        }
-    }
-
     @Override
     protected Object getField(Row row, int index) {
         Object field = super.getField(row, index);
@@ -216,8 +172,13 @@ protected Object getField(Row row, int index) {
         return field;
     }
 
+    /**
+     * 判断是否为copy模式
+     * @param insertMode
+     * @return
+     */
     private boolean checkIsCopyMode(String insertMode){
-        if(insertMode == null || insertMode.length() == 0){
+        if(StringUtils.isBlank(insertMode)){
             return false;
         }
 
@@ -227,6 +188,4 @@ private boolean checkIsCopyMode(String insertMode){
 
         return true;
     }
-
-
 }

flinkx-rdb/flinkx-rdb-core/src/main/java/com/dtstack/flinkx/rdb/util/DbUtil.java

@@ -192,6 +192,8 @@ public static void closeDbResources(ResultSet rs, Statement stmt, Connection con
             try {
                 if(commit){
                     commit(conn);
+                }else {
+                    rollBack(conn);
                 }
 
                 conn.close();
@@ -215,6 +217,20 @@ public static void commit(Connection conn){
         }
     }
 
+    /**
+     * 手动回滚事物
+     * @param conn Connection
+     */
+    public static void rollBack(Connection conn){
+        try {
+            if (!conn.isClosed() && !conn.getAutoCommit()){
+                conn.rollback();
+            }
+        } catch (SQLException e){
+            LOG.warn("rollBack error:{}", ExceptionUtil.getErrorMessage(e));
+        }
+    }
+
     /**
      * 批量执行sql
      * @param dbConn Connection

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

@@ -147,7 +147,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,6 +215,7 @@ public boolean reachedEnd() throws IOException{
     @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);
@@ -880,4 +883,19 @@ protected void checkSize(int columnCount, List<MetaColumn> metaColumns) {
             throw new RuntimeException(message);
         }
     }
+
+    /**
+     * 兼容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;
+            }
+        }
+    }
 }

flinkx-rdb/flinkx-rdb-writer/src/main/java/com/dtstack/flinkx/rdb/outputformat/JdbcOutputFormat.java

@@ -147,9 +147,8 @@ protected void openInternal(int taskNumber, int numTasks){
             ClassUtil.forName(driverName, getClass().getClassLoader());
             dbConn = DbUtil.getConnection(dbUrl, username, password);
 
-            if (restoreConfig.isRestore()){
-                dbConn.setAutoCommit(false);
-            }
+            //默认关闭事务自动提交，手动控制事务
+            dbConn.setAutoCommit(false);
 
             if(CollectionUtils.isEmpty(fullColumn)) {
                 fullColumn = probeFullColumns(table, dbConn);
@@ -180,6 +179,8 @@ protected void openInternal(int taskNumber, int numTasks){
             LOG.info("subTask[{}}] wait finished", taskNumber);
         } catch (SQLException sqe) {
             throw new IllegalArgumentException("open() failed.", sqe);
+        }finally {
+            DbUtil.commit(dbConn);
         }
     }
 
@@ -218,7 +219,9 @@ protected void writeSingleRecordInternal(Row row) throws WriteRecordException {
             }
 
             preparedStatement.execute();
+            DbUtil.commit(dbConn);
         } catch (Exception e) {
+            DbUtil.rollBack(dbConn);
             processWriteException(e, index, row);
         }
     }
@@ -231,7 +234,9 @@ protected void processWriteException(Exception e, int index, Row row) throws Wri
         }
 
         if(index < row.getArity()) {
-            throw new WriteRecordException(recordConvertDetailErrorMessage(index, row), e, index, row);
+            String message = recordConvertDetailErrorMessage(index, row);
+            LOG.error(message, e);
+            throw new WriteRecordException(message, e, index, row);
         }
         throw new WriteRecordException(e.getMessage(), e);
     }
@@ -264,17 +269,18 @@ protected void writeMultipleRecordsInternal() throws Exception {
 
             if(restoreConfig.isRestore()){
                 rowsOfCurrentTransaction += rows.size();
+            }else{
+                //手动提交事务
+                DbUtil.commit(dbConn);
             }
+            preparedStatement.clearBatch();
         } catch (Exception e){
-            if (restoreConfig.isRestore()){
-                LOG.warn("writeMultipleRecordsInternal:Start rollback");
-                dbConn.rollback();
-                LOG.warn("writeMultipleRecordsInternal:Rollback success");
-            }
             LOG.warn("write Multiple Records error, row size = {}, first row = {},  e = {}",
                     rows.size(),
                     rows.size() > 0 ? GsonUtil.GSON.toJson(rows.get(0)) : "null",
                     ExceptionUtil.getErrorMessage(e));
+            LOG.warn("error to writeMultipleRecords, start to rollback connection, e = {}", ExceptionUtil.getErrorMessage(e));
+            DbUtil.rollBack(dbConn);
             throw e;
         }finally {
             //执行完后清空batch
@@ -300,7 +306,9 @@ public FormatState getFormatState(){
                 }else{
                     preparedStatement.executeBatch();
                 }
+                //若事务提交失败，抛出异常
                 dbConn.commit();
+                preparedStatement.clearBatch();
                 LOG.info("getFormatState:Commit connection success");
 
                 snapshotWriteCounter.add(rowsOfCurrentTransaction);
@@ -321,6 +329,7 @@ public FormatState getFormatState(){
                 //执行完后清空batch
                 preparedStatement.clearBatch();
                 LOG.warn("getFormatState:Start rollback");
+                //若事务回滚失败，抛出异常
                 dbConn.rollback();
                 LOG.warn("getFormatState:Rollback success");
             } catch (SQLException sqlE){
@@ -417,6 +426,7 @@ protected boolean needWaitBeforeWriteRecords() {
 
     @Override
     protected void beforeWriteRecords()  {
+        // preSql
         if(taskNumber == 0) {
             DbUtil.executeBatch(dbConn, preSql);
         }
@@ -429,10 +439,9 @@ protected boolean needWaitBeforeCloseInternal() {
 
     @Override
     protected void beforeCloseInternal() {
-        // 执行postsql
+        // 执行postSql
         if(taskNumber == 0) {
             DbUtil.executeBatch(dbConn, postSql);
         }
     }
-
 }