Merge branch 'hotfix_1.10_4.0.x_32869' into '1.10_release_4.0.x'

[fix-32869][rdb]fix rdb task hangs after connect retry.

See merge request dt-insight-engine/flinkStreamSQL!193

Author: simenliuxing

core/src/main/java/com/dtstack/flink/sql/util/JDBCUtils.java

@@ -19,13 +19,23 @@
 
 package com.dtstack.flink.sql.util;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Objects;
 
 public class JDBCUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(JDBCUtils.class);
+
     private static final Object LOCK = new Object();
 
-    public static void forName(String clazz, ClassLoader classLoader)  {
-        synchronized (LOCK){
+    public static void forName(String clazz, ClassLoader classLoader) {
+        synchronized (LOCK) {
             try {
                 Class.forName(clazz, true, classLoader);
                 DriverManager.setLoginTimeout(10);
@@ -44,4 +54,75 @@ public synchronized static void forName(String clazz) {
             throw new RuntimeException(e);
         }
     }
+
+    /**
+     * 关闭连接资源
+     *
+     * @param rs     ResultSet
+     * @param stmt   Statement
+     * @param conn   Connection
+     * @param commit
+     */
+    public static void closeConnectionResource(ResultSet rs, Statement stmt, Connection conn, boolean commit) {
+        if (Objects.nonNull(rs)) {
+            try {
+                rs.close();
+            } catch (SQLException e) {
+                LOG.warn("Close resultSet error: {}", e.getMessage());
+            }
+        }
+
+        if (Objects.nonNull(stmt)) {
+            try {
+                stmt.close();
+            } catch (SQLException e) {
+                LOG.warn("Close statement error:{}", e.getMessage());
+            }
+        }
+
+        if (Objects.nonNull(conn)) {
+            try {
+                if (commit) {
+                    commit(conn);
+                } else {
+                    rollBack(conn);
+                }
+
+                conn.close();
+            } catch (SQLException e) {
+                LOG.warn("Close connection error:{}", e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 手动提交事物
+     *
+     * @param conn Connection
+     */
+    public static void commit(Connection conn) {
+        try {
+            if (!conn.isClosed() && !conn.getAutoCommit()) {
+                conn.commit();
+            }
+        } catch (SQLException e) {
+            LOG.warn("commit error:{}", e.getMessage());
+        }
+    }
+
+    /**
+     * 手动回滚事物
+     *
+     * @param conn Connection
+     */
+    public static void rollBack(Connection conn) {
+        try {
+            if (!conn.isClosed() && !conn.getAutoCommit()) {
+                conn.rollback();
+            }
+        } catch (SQLException e) {
+            LOG.warn("rollBack error:{}", e.getMessage());
+        }
+    }
+
 }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java

@@ -22,6 +22,7 @@
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
 import com.dtstack.flink.sql.side.rdb.util.SwitchUtil;
+import com.dtstack.flink.sql.util.JDBCUtils;
 import com.dtstack.flink.sql.util.RowDataComplete;
 import com.dtstack.flink.sql.util.RowDataConvert;
 import com.google.common.collect.Lists;
@@ -125,36 +126,26 @@ public void flatMap(Row value, Collector<BaseRow> out) throws Exception {
     }
 
     private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQLException {
-        RdbSideTableInfo tableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
-        Connection connection = null;
+        queryAndFillData(tmpCache, getConnectionWithRetry((RdbSideTableInfo) sideInfo.getSideTableInfo()));
+    }
 
-        try {
-            for (int i = 0; i < CONN_RETRY_NUM; i++) {
+    private Connection getConnectionWithRetry(RdbSideTableInfo tableInfo) throws SQLException {
+        String connInfo = "url:" + tableInfo.getUrl() + "; userName:" + tableInfo.getUserName();
+        String errorMsg = null;
+        for (int i = 0; i < CONN_RETRY_NUM; i++) {
+            try {
+                return getConn(tableInfo.getUrl(), tableInfo.getUserName(), tableInfo.getPassword());
+            } catch (Exception e) {
                 try {
-                    connection = getConn(tableInfo.getUrl(), tableInfo.getUserName(), tableInfo.getPassword());
-                    break;
-                } catch (Exception e) {
-                    if (i == CONN_RETRY_NUM - 1) {
-                        throw new RuntimeException("", e);
-                    }
-                    try {
-                        String connInfo = "url:" + tableInfo.getUrl() + ";userName:" + tableInfo.getUserName() + ",pwd:" + tableInfo.getPassword();
-                        LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
-                        Thread.sleep(5 * 1000);
-                    } catch (InterruptedException e1) {
-                        LOG.error("", e1);
-                    }
+                    LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
+                    errorMsg = e.getCause().toString();
+                    Thread.sleep(5 * 1000);
+                } catch (InterruptedException e1) {
+                    LOG.error("", e1);
                 }
             }
-            queryAndFillData(tmpCache, connection);
-        } catch (Exception e) {
-            LOG.error("", e);
-            throw new SQLException(e);
-        } finally {
-            if (connection != null) {
-                connection.close();
-            }
         }
+        throw new SQLException("get conn fail. connInfo: " + connInfo + "\ncause by: " + errorMsg);
     }
 
     private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, Connection connection) throws SQLException {
@@ -188,6 +179,7 @@ private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, C
             tmpCache.computeIfAbsent(cacheKey, key -> Lists.newArrayList())
                     .add(oneRow);
         }
+        JDBCUtils.closeConnectionResource(resultSet, statement, connection, false);
     }
 
     public int getFetchSize() {