fix last one

Author: dapeng

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

@@ -11,9 +11,8 @@ public class CustomerFlinkPartition<T> extends FlinkFixedPartitioner<T> {
     @Override
     public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
         Preconditions.checkArgument(partitions != null && partitions.length > 0, "Partitions of the target topic is empty.");
-        System.out.println("key = " + key+ ", value = " + value+ ", targetTopic = " + targetTopic + ", partitions = " + Arrays.toString(partitions));
         if(key == null){
-            super.partition(record, key, value, targetTopic, partitions);
+            return super.partition(record, key, value, targetTopic, partitions);
         }
         return partitions[key.hashCode() % partitions.length];
     }

kafka/kafka-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -77,7 +77,7 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         for (String key : kafkaSinkTableInfo.getKafkaParamKeys()) {
             properties.setProperty(key, kafkaSinkTableInfo.getKafkaParam(key));
         }
-        this.partitioner = Optional.of(getFlinkPartitioner(kafkaSinkTableInfo));
+        this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafkaSinkTableInfo);
         this.fieldNames = kafkaSinkTableInfo.getFields();
         TypeInformation[] types = new TypeInformation[kafkaSinkTableInfo.getFields().length];

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -81,7 +81,7 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
 			properties.setProperty(key, kafka09SinkTableInfo.getKafkaParam(key));
 		}
 
-		this.partitioner = Optional.of(getFlinkPartitioner(kafka09SinkTableInfo));
+		this.partitioner = Optional.of(new CustomerFlinkPartition<>());
 		this.partitionKeys = getPartitionKeys(kafka09SinkTableInfo);
 		this.fieldNames = kafka09SinkTableInfo.getFields();
 		TypeInformation[] types = new TypeInformation[kafka09SinkTableInfo.getFields().length];

kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -82,7 +82,7 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         for (String key : kafka11SinkTableInfo.getKafkaParamKeys()) {
             properties.setProperty(key, kafka11SinkTableInfo.getKafkaParam(key));
         }
-        this.partitioner = Optional.of(getFlinkPartitioner(kafka11SinkTableInfo));
+        this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafka11SinkTableInfo);
         this.fieldNames = kafka11SinkTableInfo.getFields();
         TypeInformation[] types = new TypeInformation[kafka11SinkTableInfo.getFields().length];