[hotfix-35512][rdb] adjust code.

Author: FlechazoW

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

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.core.rdb;
 
-import com.dtstack.flink.sql.core.rdb.util.JdbcConnectUtil;
+import com.dtstack.flink.sql.core.rdb.util.JdbcConnectionUtil;
 import com.dtstack.flink.sql.resource.ResourceCheck;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.runtime.execution.SuppressRestartsException;
@@ -109,7 +109,7 @@ public void checkPrivilege(
             , String schema
             , List<String> privilegeList) {
         Connection connection =
-                JdbcConnectUtil.getConnectWithRetry(driverName, url, userName, password);
+                JdbcConnectionUtil.getConnectionWithRetry(driverName, url, userName, password);
         Statement statement = null;
         String tableInfo = Objects.isNull(schema) ? tableName : schema + "." + tableName;
         String privilege = null;
@@ -133,7 +133,7 @@ public void checkPrivilege(
 
             throw new SuppressRestartsException(new IllegalArgumentException(sqlException.getMessage()));
         } finally {
-            JdbcConnectUtil.closeConnectionResource(null, statement, connection, false);
+            JdbcConnectionUtil.closeConnectionResource(null, statement, connection, false);
         }
     }
 }

…k/sql/core/rdb/util/JdbcConnectUtil.java …ql/core/rdb/util/JdbcConnectionUtil.javardb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectUtil.java renamed to rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectionUtil.java rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectUtil.java renamed to rdb/rdb-core/src/main/java/com/dtstack/flink/sql/core/rdb/util/JdbcConnectionUtil.java

@@ -38,11 +38,11 @@
  * Date 2020-12-25
  * Company dtstack
  */
-public class JdbcConnectUtil {
+public class JdbcConnectionUtil {
     private static final int DEFAULT_RETRY_NUM = 3;
     private static final long DEFAULT_RETRY_TIME_WAIT = 3L;
     private static final int DEFAULT_VALID_TIME = 10;
-    private static final Logger LOG = LoggerFactory.getLogger(JdbcConnectUtil.class);
+    private static final Logger LOG = LoggerFactory.getLogger(JdbcConnectionUtil.class);
 
     /**
      * 关闭连接资源
@@ -119,25 +119,26 @@ public static void rollBack(Connection conn) {
     }
 
     /**
-     * get connect from datasource and retry when failed.
+     * get connection from datasource and retry when failed.
      *
      * @param driverName driver name for rdb datasource
      * @param url        connect url
      * @param userName   connect user name
      * @param password   password for user name
      * @return a valid connection
      */
-    public static Connection getConnectWithRetry(
-            String driverName
-            , String url
-            , String userName
-            , String password) {
-        String errorMessage = "\nGet connect failed with properties: \nurl: " + url
-                + (Objects.isNull(userName) ? "" : "\nuserName: " + userName
-                + "\nerror message: ");
-        String errorCause = null;
+    public static Connection getConnectionWithRetry(String driverName,
+                                                    String url,
+                                                    String userName,
+                                                    String password) {
+        String message = "Get connection failed. " +
+            "\nurl: [%s]" +
+            "\nuserName: [%s]" +
+            "\ncause: [%s]";
+        String errorCause;
+        String errorMessage = "";
 
-        ClassLoaderManager.forName(driverName, JdbcConnectUtil.class.getClassLoader());
+        ClassLoaderManager.forName(driverName, JdbcConnectionUtil.class.getClassLoader());
         Preconditions.checkNotNull(url, "url can't be null!");
 
         for (int i = 0; i < DEFAULT_RETRY_NUM; i++) {
@@ -150,11 +151,17 @@ public static Connection getConnectWithRetry(
                 return connection;
             } catch (Exception e) {
                 errorCause = ExceptionTrace.traceOriginalCause(e);
-                LOG.warn(errorMessage + errorCause);
+                errorMessage = String.format(
+                    message,
+                    url,
+                    userName,
+                    errorCause
+                );
+                LOG.warn(errorMessage);
                 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 SQLException(errorMessage + errorCause));
+        throw new SuppressRestartsException(new SQLException(errorMessage));
     }
 }

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

@@ -19,7 +19,7 @@
 package com.dtstack.flink.sql.side.rdb.all;
 
 import com.dtstack.flink.sql.core.rdb.JdbcResourceCheck;
-import com.dtstack.flink.sql.core.rdb.util.JdbcConnectUtil;
+import com.dtstack.flink.sql.core.rdb.util.JdbcConnectionUtil;
 import com.dtstack.flink.sql.side.BaseAllReqRow;
 import com.dtstack.flink.sql.side.BaseSideInfo;
 import com.dtstack.flink.sql.side.rdb.table.RdbSideTableInfo;
@@ -208,7 +208,7 @@ private void queryAndFillData(Map<String, List<Map<String, Object>>> tmpCache, C
             tmpCache.computeIfAbsent(cacheKey, key -> Lists.newArrayList())
                     .add(oneRow);
         }
-        JdbcConnectUtil.closeConnectionResource(resultSet, statement, connection, false);
+        JdbcConnectionUtil.closeConnectionResource(resultSet, statement, connection, false);
     }
 
     public int getFetchSize() {

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

@@ -20,7 +20,7 @@
 
 
 import com.dtstack.flink.sql.core.rdb.JdbcResourceCheck;
-import com.dtstack.flink.sql.core.rdb.util.JdbcConnectUtil;
+import com.dtstack.flink.sql.core.rdb.util.JdbcConnectionUtil;
 import com.dtstack.flink.sql.enums.EUpdateMode;
 import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.factory.DTThreadFactory;
@@ -135,7 +135,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
 
     public void openJdbc() throws IOException {
         try {
-            connection = JdbcConnectUtil.getConnectWithRetry(
+            connection = JdbcConnectionUtil.getConnectionWithRetry(
                 driverName,
                 dbURL,
                 username,
@@ -194,7 +194,7 @@ private void checkConnectionOpen() {
         try {
             if (!connection.isValid(10)) {
                 LOG.info("db connection reconnect..");
-                connection = JdbcConnectUtil.getConnectWithRetry(
+                connection = JdbcConnectionUtil.getConnectionWithRetry(
                     driverName,
                     dbURL,
                     username,

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/JDBCWriter.java

@@ -18,7 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb.writer;
 
-import com.dtstack.flink.sql.core.rdb.util.JdbcConnectUtil;
+import com.dtstack.flink.sql.core.rdb.util.JdbcConnectionUtil;
 import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.outputformat.AbstractDtRichOutputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -82,8 +82,8 @@ default void dealExecuteError(Connection connection,
 								  Row row,
 								  long errorLimit,
 								  Logger LOG) {
-		JdbcConnectUtil.rollBack(connection);
-		JdbcConnectUtil.commit(connection);
+		JdbcConnectionUtil.rollBack(connection);
+		JdbcConnectionUtil.commit(connection);
 
 		if (metricOutputFormat.outDirtyRecords.getCount() % DIRTYDATA_PRINT_FREQUENTY == 0 ||
 			LOG.isDebugEnabled()) {