Merge branch 'v1.5.0_dev' of github.com:DTStack/flinkStreamSQL into v1.5.0_dev

Author: yanxi0227

docs/kafkaSource.md

@@ -39,7 +39,7 @@ CREATE TABLE tableName(
 |zookeeperQuorum | kafka zk地址信息(多个之间用逗号分隔)|是||
 |topic | 需要读取的 topic 名称|是||
 |topicIsPattern |  topic是否是正则表达式格式|否| false
-|groupId | 需要读取的 groupId 名称|是||
+|groupId | 需要读取的 groupId 名称|否||
 |offsetReset | 读取的topic 的offset初始位置[latest\|earliest\|指定offset值({"0":12312,"1":12321,"2":12312},{"partition_no":offset_value})]|否|latest|
 |parallelism | 并行度设置|否|1|
 

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

@@ -69,7 +69,9 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
         Properties props = new Properties();
         props.setProperty("bootstrap.servers", kafka09SourceTableInfo.getBootstrapServers());
         props.setProperty("auto.offset.reset", kafka09SourceTableInfo.getOffsetReset());
-        props.setProperty("group.id", kafka09SourceTableInfo.getGroupId());
+        if (StringUtils.isNotBlank(kafka09SourceTableInfo.getGroupId())){
+            props.setProperty("group.id", kafka09SourceTableInfo.getGroupId());
+        }
         // only required for Kafka 0.8
         //TODO props.setProperty("zookeeper.connect", kafka09SourceTableInfo.)
 

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

@@ -121,7 +121,7 @@ public void setOffset(String offset) {
 	public boolean check() {
 		Preconditions.checkNotNull(bootstrapServers, "kafka of bootstrapServers is required");
 		Preconditions.checkNotNull(topic, "kafka of topic is required");
-		Preconditions.checkNotNull(groupId, "kafka of groupId is required");
+		//Preconditions.checkNotNull(groupId, "kafka of groupId is required");
 		Preconditions.checkState(offsetReset.equalsIgnoreCase("latest")
 				|| offsetReset.equalsIgnoreCase("latest"), "kafka of offsetReset set fail");
 

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

@@ -69,7 +69,9 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 		Properties props = new Properties();
 		props.setProperty("bootstrap.servers", kafka010SourceTableInfo.getBootstrapServers());
 		props.setProperty("auto.offset.reset", kafka010SourceTableInfo.getOffsetReset());
-		props.setProperty("group.id", kafka010SourceTableInfo.getGroupId());
+		if (StringUtils.isNotBlank(kafka010SourceTableInfo.getGroupId())){
+			props.setProperty("group.id", kafka010SourceTableInfo.getGroupId());
+		}
 		// only required for Kafka 0.8
 		//TODO props.setProperty("zookeeper.connect", kafka09SourceTableInfo.)
 

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

@@ -121,7 +121,7 @@ public void setTopicIsPattern(Boolean topicIsPattern) {
     public boolean check() {
         Preconditions.checkNotNull(bootstrapServers, "kafka of bootstrapServers is required");
         Preconditions.checkNotNull(topic, "kafka of topic is required");
-        Preconditions.checkNotNull(groupId, "kafka of groupId is required");
+        //Preconditions.checkNotNull(groupId, "kafka of groupId is required");
         Preconditions.checkState(offsetReset.equalsIgnoreCase("latest")
                 || offsetReset.equalsIgnoreCase("latest"), "kafka of offsetReset set fail");
 

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

@@ -69,7 +69,9 @@ public Table genStreamSource(SourceTableInfo sourceTableInfo, StreamExecutionEnv
 		Properties props = new Properties();
 		props.setProperty("bootstrap.servers", kafka011SourceTableInfo.getBootstrapServers());
 		props.setProperty("auto.offset.reset", kafka011SourceTableInfo.getOffsetReset());
-		props.setProperty("group.id", kafka011SourceTableInfo.getGroupId());
+		if (StringUtils.isNotBlank(kafka011SourceTableInfo.getGroupId())){
+			props.setProperty("group.id", kafka011SourceTableInfo.getGroupId());
+		}
 		// only required for Kafka 0.8
 		//TODO props.setProperty("zookeeper.connect", kafka09SourceTableInfo.)
 

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

@@ -122,7 +122,7 @@ public void setOffset(String offset) {
 	public boolean check() {
 		Preconditions.checkNotNull(bootstrapServers, "kafka of bootstrapServers is required");
 		Preconditions.checkNotNull(topic, "kafka of topic is required");
-		Preconditions.checkNotNull(groupId, "kafka of groupId is required");
+		//Preconditions.checkNotNull(groupId, "kafka of groupId is required");
 		Preconditions.checkState(offsetReset.equalsIgnoreCase("latest")
 				|| offsetReset.equalsIgnoreCase("latest"), "kafka of offsetReset set fail");