[hotfix-35526][rdb]调整rdb获取连接方法调用，统一走JdbcConnectUtil的getConnectWithRetry

Author: FlechazoW

rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectUtil.java

@@ -19,6 +19,7 @@
 package com.dtstack.flink.sql.core.rdb.util;
 
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
+import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.util.ThreadUtil;
 import com.google.common.base.Preconditions;
 import org.apache.flink.runtime.execution.SuppressRestartsException;
@@ -141,20 +142,19 @@ public static Connection getConnectWithRetry(
 
         for (int i = 0; i < DEFAULT_RETRY_NUM; i++) {
             try {
-                return Objects.isNull(userName) ?
-                        DriverManager.getConnection(url) : DriverManager.getConnection(url, userName, password);
+                Connection connection =
+                    Objects.isNull(userName) ?
+                        DriverManager.getConnection(url) :
+                        DriverManager.getConnection(url, userName, password);
+                connection.setAutoCommit(false);
+                return connection;
             } catch (Exception e) {
-                if (Objects.isNull(e.getCause())) {
-                    errorCause = e.getMessage();
-                } else {
-                    errorCause = e.getCause().toString();
-                }
-
+                errorCause = ExceptionTrace.traceOriginalCause(e);
                 LOG.warn(errorMessage + errorCause);
                 LOG.warn("Connect will retry after [{}] s. Retry time [{}] ...", DEFAULT_RETRY_TIME_WAIT, i + 1);
                 ThreadUtil.sleepSeconds(DEFAULT_RETRY_TIME_WAIT);
             }
         }
-        throw new SuppressRestartsException(new Throwable(errorMessage + errorCause));
+        throw new SuppressRestartsException(new SQLException(errorMessage + errorCause));
     }
 }

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

@@ -18,7 +18,6 @@
 
 package com.dtstack.flink.sql.sink.rdb.format;
 
-import com.dtstack.flink.sql.classloader.ClassLoaderManager;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.types.Row;
@@ -63,16 +62,6 @@ public AbstractJDBCOutputFormat(String username, String password, String driverN
 	public void configure(Configuration parameters) {
 	}
 
-	protected void establishConnection() throws SQLException, ClassNotFoundException, IOException {
-		ClassLoaderManager.forName(driverName, getClass().getClassLoader());
-		if (username == null) {
-			connection = DriverManager.getConnection(dbURL);
-		} else {
-			connection = DriverManager.getConnection(dbURL, username, password);
-		}
-		connection.setAutoCommit(false);
-	}
-
 	protected void closeDbConnection() throws IOException {
 		if (connection != null) {
 			try {

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

@@ -20,6 +20,7 @@
 
 
 import com.dtstack.flink.sql.core.rdb.JdbcResourceCheck;
+import com.dtstack.flink.sql.core.rdb.util.JdbcConnectUtil;
 import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.factory.DTThreadFactory;
@@ -134,7 +135,12 @@ public void open(int taskNumber, int numTasks) throws IOException {
 
     public void openJdbc() throws IOException {
         try {
-            establishConnection();
+            connection = JdbcConnectUtil.getConnectWithRetry(
+                driverName,
+                dbURL,
+                username,
+                password
+            );
             initMetric();
             if(jdbcWriter == null){
                 if (StringUtils.equalsIgnoreCase(updateMode, EUpdateMode.APPEND.name()) || keyFields == null || keyFields.length == 0) {
@@ -152,8 +158,6 @@ public void openJdbc() throws IOException {
             jdbcWriter.open(connection);
         } catch (SQLException sqe) {
             throw new IllegalArgumentException("open() failed.", sqe);
-        } catch (ClassNotFoundException cnfe) {
-            throw new IllegalArgumentException("JDBC driver class not found.", cnfe);
         }
 
         if (flushIntervalMills != 0) {
@@ -190,15 +194,16 @@ private void checkConnectionOpen() {
         try {
             if (!connection.isValid(10)) {
                 LOG.info("db connection reconnect..");
-                establishConnection();
+                connection = JdbcConnectUtil.getConnectWithRetry(
+                    driverName,
+                    dbURL,
+                    username,
+                    password
+                );
                 jdbcWriter.prepareStatement(connection);
             }
         } catch (SQLException e) {
             LOG.error("check connection open failed..", e);
-        } catch (ClassNotFoundException e) {
-            LOG.error("load jdbc class error when reconnect db..", e);
-        } catch (IOException e) {
-            LOG.error("jdbc io exception..", e);
         }
     }
 
@@ -208,7 +213,7 @@ public synchronized void flush() {
             batchCount = 0;
         } catch (Exception e) {
             String errorMsg = String.format(
-                "Writing records to JDBC failed. %s",
+                "Writing records to JDBC failed. Cause: [%s]",
                 ExceptionTrace.traceOriginalCause(e));
 
             ExceptionTrace.dealExceptionWithSuppressStart(