[flinkStremSql][415]

1.支持elasticsearch7 sink以及side，并完善其文档
2.支持脏数据管理
3.支持file source
4.impla插件批量写入失败时改为单条insert
5.支持http sink

Author: HiLany

README.md

@@ -41,5 +41,7 @@ FlinkStreamSQL
 
 ## 如何贡献FlinkStreamSQL
 
+[pr规范](docs/pr.md)
+
 ## License
 FlinkStreamSQL is under the Apache 2.0 license. See the [LICENSE](http://www.apache.org/licenses/LICENSE-2.0) file for details.    

docs/pr.md

@@ -0,0 +1,20 @@
+## PR规范
+
+1. 建立issue,描述相关问题信息
+2. 基于对应的release分支拉取开发分支
+3. commit 信息：[type-issueid] [module] msg
+   1. type 类别
+   2. feat：表示是一个新功能（feature)
+   3. hotfix：hotfix，修补bug
+   4. docs：改动、增加文档
+   5. opt：修改代码风格及opt imports这些，不改动原有执行的代码
+   6. test：增加测试
+
+<br/>  
+
+eg:
+  [hotfix-31280][core] 修复bigdecimal转decimal运行失败问题               
+  [feat-31372][rdb] RDB结果表Upsert模式支持选择更新策略
+  
+1. 多次提交使用rebase 合并成一个。
+2. pr 名称：[flinkx-issueid][module名称] 标题

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.getConnectWithRetry(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);
         }
     }
 }

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);
 
     /**
      * 关闭连接资源
@@ -137,7 +137,7 @@ public static Connection getConnectWithRetry(
                 + "\nerror message: ");
         String errorCause = null;
 
-        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++) {

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.getConnectWithRetry(
                 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.getConnectWithRetry(
                     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()) {