kafkasource parallelism

todd5167 · todd5167 · commit 8882e8d39b3b · 2020-03-26T17:39:43.000+08:00
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -112,7 +112,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
         DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
         Integer parallelism = kafkaSourceTableInfo.getParallelism();
-        if (parallelism != null) {
+        if (parallelism > 0) {
             kafkaSource.setParallelism(parallelism);
         }
         return tableEnv.fromDataStream(kafkaSource, fields);
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -112,7 +112,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
 		DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 		Integer parallelism = kafkaSourceTableInfo.getParallelism();
-		if (parallelism != null) {
+		if (parallelism > 0) {
 			kafkaSource.setParallelism(parallelism);
 		}
 		return tableEnv.fromDataStream(kafkaSource, fields);
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/KafkaSource.java
@@ -115,7 +115,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec
 
 		DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);
 		Integer parallelism = kafkaSourceTableInfo.getParallelism();
-		if (parallelism != null) {
+		if (parallelism > 0) {
 			kafkaSource.setParallelism(parallelism);
 		}
 		return tableEnv.fromDataStream(kafkaSource, fields);

Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec`
`112`	`112`
`113`	`113`	`DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);`
`114`	`114`	`Integer parallelism = kafkaSourceTableInfo.getParallelism();`
`115`		`- if (parallelism != null) {`
	`115`	`+ if (parallelism > 0) {`
`116`	`116`	`kafkaSource.setParallelism(parallelism);`
`117`	`117`	`}`
`118`	`118`	`return tableEnv.fromDataStream(kafkaSource, fields);`
Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,7 @@ public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo, StreamExec`
`115`	`115`
`116`	`116`	`DataStreamSource kafkaSource = env.addSource(kafkaSrc, sourceOperatorName, typeInformation);`
`117`	`117`	`Integer parallelism = kafkaSourceTableInfo.getParallelism();`
`118`		`- if (parallelism != null) {`
	`118`	`+ if (parallelism > 0) {`
`119`	`119`	`kafkaSource.setParallelism(parallelism);`
`120`	`120`	`}`
`121`	`121`	`return tableEnv.fromDataStream(kafkaSource, fields);`