Merge branch '1.5_v3.5.1_check_kafkalist' into '1.5_v3.5.1'

kafka brokerlist 是空的不允许运行和提交



See merge request !2

Author: toutian

core/src/main/java/com/dtstack/flink/sql/table/AbsTableParser.java

@@ -56,7 +56,7 @@ protected boolean fieldNameNeedsUpperCase() {
         return true;
     }
 
-    public abstract TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props);
+    public abstract TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) throws Exception;
 
     public boolean dealKeyPattern(String fieldRow, TableInfo tableInfo){
         for(Map.Entry<String, Pattern> keyPattern : keyPatternMap.entrySet()){

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

@@ -62,14 +62,19 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
     }
 
     @Override
-    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) throws Exception {
 
         KafkaSourceTableInfo kafka09SourceTableInfo = new KafkaSourceTableInfo();
         kafka09SourceTableInfo.setName(tableName);
         parseFieldsInfo(fieldsInfo, kafka09SourceTableInfo);
 
         kafka09SourceTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(KafkaSourceTableInfo.PARALLELISM_KEY.toLowerCase())));
-        kafka09SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
+        String bootstrapServer = MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase()));
+        if (bootstrapServer == null || bootstrapServer.trim().equals("")){
+            throw new Exception("BootstrapServers can not be empty!");
+        } else {
+            kafka09SourceTableInfo.setBootstrapServers(bootstrapServer);
+        }
         kafka09SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka09SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka09SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));

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

@@ -66,14 +66,20 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
     }
 
     @Override
-    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) throws Exception {
 
         KafkaSourceTableInfo kafka10SourceTableInfo = new KafkaSourceTableInfo();
         kafka10SourceTableInfo.setName(tableName);
         parseFieldsInfo(fieldsInfo, kafka10SourceTableInfo);
 
         kafka10SourceTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(KafkaSourceTableInfo.PARALLELISM_KEY.toLowerCase())));
-        kafka10SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
+
+        String bootstrapServer = MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase()));
+        if (bootstrapServer == null || bootstrapServer.trim().equals("")){
+            throw new Exception("BootstrapServers can not be empty!");
+        } else {
+            kafka10SourceTableInfo.setBootstrapServers(bootstrapServer);
+        }
         kafka10SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka10SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka10SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));

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

@@ -66,14 +66,19 @@ static void dealNestField(Matcher matcher, TableInfo tableInfo) {
     }
 
     @Override
-    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) {
+    public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, Object> props) throws Exception {
 
         KafkaSourceTableInfo kafka11SourceTableInfo = new KafkaSourceTableInfo();
         kafka11SourceTableInfo.setName(tableName);
         parseFieldsInfo(fieldsInfo, kafka11SourceTableInfo);
 
         kafka11SourceTableInfo.setParallelism(MathUtil.getIntegerVal(props.get(KafkaSourceTableInfo.PARALLELISM_KEY.toLowerCase())));
-        kafka11SourceTableInfo.setBootstrapServers(MathUtil.getString(props.get(KafkaSourceTableInfo.BOOTSTRAPSERVERS_KEY.toLowerCase())));
+        String bootstrapServer = MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase()));
+        if (bootstrapServer == null || bootstrapServer.trim().equals("")){
+            throw new Exception("BootstrapServers can not be empty!");
+        } else {
+            kafka11SourceTableInfo.setBootstrapServers(bootstrapServer);
+        }
         kafka11SourceTableInfo.setGroupId(MathUtil.getString(props.get(KafkaSourceTableInfo.GROUPID_KEY.toLowerCase())));
         kafka11SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka11SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));