kafka source parallelsim

Author: todd5167

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

@@ -26,6 +26,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.table.api.Table;
@@ -79,6 +80,12 @@ protected TypeInformation<Row> getRowTypeInformation(KafkaSourceTableInfo kafkaS
         return new RowTypeInfo(types, kafkaSourceTableInfo.getFields());
     }
 
+    protected void setParallelism(Integer parallelism, DataStreamSource kafkaSource) {
+        if (parallelism > 0) {
+            kafkaSource.setParallelism(parallelism);
+        }
+    }
+
     protected void setStartPosition(String offset, String topicName, FlinkKafkaConsumerBase<Row> kafkaSrc) {
         if (StringUtils.equalsIgnoreCase(offset, EKafkaOffset.EARLIEST.name())) {
             kafkaSrc.setStartFromEarliest();

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

@@ -49,8 +49,8 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
         String sourceOperatorName = generateOperatorName(sourceTableInfo.getName(), topicName);
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
-        kafkaSource.setParallelism(kafkaSourceTableInfo.getParallelism());
 
+        setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
         setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 

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

@@ -21,18 +21,14 @@
 
 import com.dtstack.flink.sql.source.kafka.table.KafkaSourceTableInfo;
 import com.dtstack.flink.sql.table.AbstractSourceTableInfo;
-import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
-import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.types.Row;
-
-import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -54,11 +50,13 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
         String sourceOperatorName = generateOperatorName(sourceTableInfo.getName(), topicName);
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
-        kafkaSource.setParallelism(kafkaSourceTableInfo.getParallelism());
 
+        setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
         setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 
         return tableEnv.fromDataStream(kafkaSource, fields);
     }
+
+
 }

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

@@ -53,8 +53,8 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
         String sourceOperatorName = generateOperatorName(sourceTableInfo.getName(), topicName);
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
-        kafkaSource.setParallelism(kafkaSourceTableInfo.getParallelism());
 
+        setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
         setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");
 

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

@@ -52,8 +52,8 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
         String sourceOperatorName = generateOperatorName(sourceTableInfo.getName(), topicName);
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
-        kafkaSource.setParallelism(kafkaSourceTableInfo.getParallelism());
 
+        setParallelism(kafkaSourceTableInfo.getParallelism(), kafkaSource);
         setStartPosition(kafkaSourceTableInfo.getOffsetReset(), topicName, kafkaSrc);
         String fields = StringUtils.join(kafkaSourceTableInfo.getFields(), ",");