Merge remote-tracking branch 'origin/feat_1.8_codereview' into v1.8.0_dev

Author: todd5167

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

@@ -266,7 +266,7 @@ public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrl
      */
     public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String localSqlPluginPath,
                                          String remoteSqlPluginPath, String pluginLoadMode, Map<String, AbstractSideTableInfo> sideTableMap, Map<String, Table> registerTableCache) throws Exception {
-        Set<URL> pluginClassPatshSets = Sets.newHashSet();
+        Set<URL> pluginClassPathSets = Sets.newHashSet();
         WaterMarkerAssigner waterMarkerAssigner = new WaterMarkerAssigner();
         for (AbstractTableInfo tableInfo : sqlTree.getTableInfoMap().values()) {
 
@@ -302,26 +302,26 @@ public static Set<URL> registerTable(SqlTree sqlTree, StreamExecutionEnvironment
                 registerTableCache.put(tableInfo.getName(), regTable);
 
                 URL sourceTablePathUrl = PluginUtil.buildSourceAndSinkPathByLoadMode(tableInfo.getType(), AbstractSourceTableInfo.SOURCE_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
-                pluginClassPatshSets.add(sourceTablePathUrl);
+                pluginClassPathSets.add(sourceTablePathUrl);
             } else if (tableInfo instanceof AbstractTargetTableInfo) {
 
                 TableSink tableSink = StreamSinkFactory.getTableSink((AbstractTargetTableInfo) tableInfo, localSqlPluginPath);
                 TypeInformation[] flinkTypes = FunctionManager.transformTypes(tableInfo.getFieldClasses());
                 tableEnv.registerTableSink(tableInfo.getName(), tableInfo.getFields(), flinkTypes, tableSink);
 
                 URL sinkTablePathUrl = PluginUtil.buildSourceAndSinkPathByLoadMode(tableInfo.getType(), AbstractTargetTableInfo.TARGET_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
-                pluginClassPatshSets.add(sinkTablePathUrl);
+                pluginClassPathSets.add(sinkTablePathUrl);
             } else if (tableInfo instanceof AbstractSideTableInfo) {
                 String sideOperator = ECacheType.ALL.name().equals(((AbstractSideTableInfo) tableInfo).getCacheType()) ? "all" : "async";
                 sideTableMap.put(tableInfo.getName(), (AbstractSideTableInfo) tableInfo);
 
                 URL sideTablePathUrl = PluginUtil.buildSidePathByLoadMode(tableInfo.getType(), sideOperator, AbstractSideTableInfo.TARGET_SUFFIX, localSqlPluginPath, remoteSqlPluginPath, pluginLoadMode);
-                pluginClassPatshSets.add(sideTablePathUrl);
+                pluginClassPathSets.add(sideTablePathUrl);
             } else {
                 throw new RuntimeException("not support table type:" + tableInfo.getType());
             }
         }
-        return pluginClassPatshSets;
+        return pluginClassPathSets;
     }
 
     /**

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkTableInfo.java

@@ -58,7 +58,7 @@ public class KafkaSinkTableInfo extends AbstractTargetTableInfo {
 
     private String bootstrapServers;
 
-    public Map<String, String> kafkaParam = new HashMap<String, String>();
+    public Map<String, String> kafkaParams = new HashMap<String, String>();
 
     private String topic;
 
@@ -73,15 +73,15 @@ public class KafkaSinkTableInfo extends AbstractTargetTableInfo {
     private String updateMode;
 
     public void addKafkaParam(String key, String value) {
-        kafkaParam.put(key, value);
+        kafkaParams.put(key, value);
     }
 
     public String getKafkaParam(String key) {
-        return kafkaParam.get(key);
+        return kafkaParams.get(key);
     }
 
     public Set<String> getKafkaParamKeys() {
-        return kafkaParam.keySet();
+        return kafkaParams.keySet();
     }
 
     public String getBootstrapServers() {

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

@@ -69,7 +69,7 @@ public class KafkaSourceTableInfo extends AbstractSourceTableInfo {
 
     private String fieldDelimiter;
 
-    public Map<String, String> kafkaParam = new HashMap<>();
+    public Map<String, String> kafkaParams = new HashMap<>();
 
 
     public String getBootstrapServers() {
@@ -113,15 +113,15 @@ public void setTopicIsPattern(Boolean topicIsPattern) {
     }
 
     public void addKafkaParam(Map<String, String> kafkaParam) {
-        kafkaParam.putAll(kafkaParam);
+        kafkaParams.putAll(kafkaParam);
     }
 
     public String getKafkaParam(String key) {
-        return kafkaParam.get(key);
+        return kafkaParams.get(key);
     }
 
     public Set<String> getKafkaParamKeys() {
-        return kafkaParam.keySet();
+        return kafkaParams.keySet();
     }
 
     public String getSourceDataType() {