[fix] sink add parallelism

Author: FlechazoW

kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/table/KafkaSinkParser.java

@@ -57,7 +57,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSinkTableInfo.setPartitionKeys(MathUtil.getString(props.get(KafkaSinkTableInfo.PARTITION_KEY.toLowerCase())));
         kafkaSinkTableInfo.setUpdateMode(MathUtil.getString(props.getOrDefault(KafkaSinkTableInfo.UPDATE_KEY.toLowerCase(), EUpdateMode.APPEND.name())));
 
-        Integer parallelism = MathUtil.getIntegerVal(props.get(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase()));
+        Integer parallelism = MathUtil.getIntegerVal(props.getOrDefault(KafkaSinkTableInfo.PARALLELISM_KEY.toLowerCase(), 1));
         kafkaSinkTableInfo.setParallelism(parallelism);
 
         for (String key : props.keySet()) {

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduSink.java

@@ -17,6 +17,7 @@
 import org.apache.flink.types.Row;
 
 import java.io.Serializable;
+import java.util.Objects;
 
 public class KuduSink implements RetractStreamTableSink<Row>, Serializable, IStreamSinkGener<KuduSink> {
 
@@ -36,7 +37,7 @@ public class KuduSink implements RetractStreamTableSink<Row>, Serializable, IStr
 
     private Integer defaultSocketReadTimeoutMs;
 
-    private int parallelism = -1;
+    private int parallelism = 1;
 
     private String principal;
     private String keytab;
@@ -56,7 +57,8 @@ public KuduSink genStreamSink(AbstractTargetTableInfo targetTableInfo) {
         this.keytab = kuduTableInfo.getKeytab();
         this.krb5conf = kuduTableInfo.getKrb5conf();
         this.enableKrb = kuduTableInfo.isEnableKrb();
-        this.parallelism = kuduTableInfo.getParallelism();
+        this.parallelism = Objects.isNull(kuduTableInfo.getParallelism()) ?
+                parallelism : kuduTableInfo.getParallelism();
 
         return this;
     }