Merge remote-tracking branch 'origin/1.8_release_3.10.x' into feat_1.8_merge3.10.x

# Conflicts:
#	README.md
#	core/pom.xml
#	core/src/main/java/com/dtstack/flink/sql/Main.java
#	core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java
#	core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java
#	core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java
#	core/src/main/java/com/dtstack/flink/sql/exec/ParamsInfo.java
#	elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/ElasticsearchSink.java
#	kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/environment/StreamEnvConfigManager.java

@@ -18,12 +18,6 @@
 
 package com.dtstack.flink.sql.environment;
 
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
-import com.dtstack.flink.sql.enums.EStateBackend;
-import com.dtstack.flink.sql.util.MathUtil;
-import com.dtstack.flink.sql.util.PropertiesUtils;
-import org.apache.commons.lang3.BooleanUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -206,7 +206,6 @@ private static void sqlTranslation(String localSqlPluginPath,
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
-
             boolean isSide = false;
             for (String tableName : result.getTargetTableList()) {
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
@@ -233,7 +232,6 @@ private static void sqlTranslation(String localSqlPluginPath,
                         System.out.println(result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if (LOG.isInfoEnabled()) {
-                            System.out.println();
                             LOG.info("exec sql: " + result.getExecSql());
                         }
                     }
@@ -355,6 +353,7 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         return env;
     }
 
+
     public static void setLogLevel(ParamsInfo paramsInfo){
         String logLevel = paramsInfo.getConfProp().getProperty(ConfigConstrant.LOG_LEVEL_KEY);
         if(org.apache.commons.lang3.StringUtils.isBlank(logLevel)){

core/src/main/java/com/dtstack/flink/sql/side/BaseAllReqRow.java

@@ -20,14 +20,15 @@
 
 package com.dtstack.flink.sql.side;
 
-import com.dtstack.flink.sql.factory.DTThreadFactory;
-import org.apache.calcite.sql.JoinType;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.runtime.types.CRow;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;
 
+import com.dtstack.flink.sql.factory.DTThreadFactory;
+import org.apache.calcite.sql.JoinType;
+
 import java.sql.SQLException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -793,9 +793,9 @@ private void joinFun(Object pollObj, Map<String, Table> localTableCache,
             throw new RuntimeException("can't not find side table:" + joinInfo.getRightTableName());
         }
 
-        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo)){
-            throw new RuntimeException("ON condition must contain all equal fields!!!");
-        }
+//        if(!checkJoinCondition(joinInfo.getCondition(), joinInfo.getRightTableAlias(), sideTableInfo)){
+//            throw new RuntimeException("ON condition must contain all equal fields!!!");
+//        }
 
         rightScopeChild.setRowTypeInfo(sideTableInfo.getRowTypeInfo());
 

docs/elasticsearch6Sink.md

@@ -35,7 +35,7 @@ CREATE TABLE tableName(
 |cluster | ES 集群名称 |是||
 |index | 选择的ES上的index名称|是||
 |esType | 选择ES上的type名称|是||
-|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id;|是||
+|id | 生成id的规则(当前是根据指定的字段pos获取字段信息,拼接生成id;|否||
 | |若id为空字符串或索引都超出范围，则随机生成id值)|||
 |authMesh | 是否进行用户名密码认证 | 否 | false|
 |userName | 用户名 | 否，authMesh='true'时为必填 ||

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -51,7 +51,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
-        kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));
+        kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase()), false));
         kafkaSourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
         for (String key : props.keySet()) {
             if (!key.isEmpty() && key.startsWith("kafka.")) {