groupid check null

Author: todd5167

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.)
 

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.)
 

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.)
 

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/ExtendOutputFormat.java

@@ -70,7 +70,7 @@ public void fillRealIndexes() throws SQLException {
             }
             String column_name = rs.getString("COLUMN_NAME");
             if (StringUtils.isNotBlank(column_name)) {
-                column_name = column_name.toUpperCase();
+                column_name = column_name;
             }
             map.get(indexName).add(column_name);
         }
@@ -94,7 +94,7 @@ public void fillFullColumns() throws SQLException {
         while (rs.next()) {
             String columnName = rs.getString("COLUMN_NAME");
             if (StringUtils.isNotBlank(columnName)) {
-                getFullField().add(columnName.toUpperCase());
+                getFullField().add(columnName);
             }
         }
     }