so

Author: dapeng

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

@@ -14,8 +14,6 @@ public int partition(T record, byte[] key, byte[] value, String targetTopic, int
         if(key == null){
             return super.partition(record, key, value, targetTopic, partitions);
         }
-        System.out.println("hashcode=" + key.hashCode());
-        Random random = new Random();
-        return partitions[random.nextInt() % partitions.length];
+        return partitions[key.hashCode() % partitions.length];
     }
 }

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

@@ -51,7 +51,6 @@ private byte[] serializeJsonKey(JsonRowSerializationSchema jsonRowSerializationS
                     sb.append(objectNode.get(key));
                 }
             }
-            System.out.println("serialKey=" + sb.toString());
             return sb.toString().getBytes();
         }catch (Exception e){
 

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

@@ -40,7 +40,7 @@ public class KafkaProducer010 extends FlinkKafkaProducer010<Row> {
     private SerializationMetricWrapper serializationMetricWrapper;
 
     public KafkaProducer010(String topicId, SerializationSchema<Row> serializationSchema, Properties producerConfig, Optional<FlinkKafkaPartitioner<Row>> customPartitioner, String[] partitionKeys) {
-        super(topicId, new CustomerKeyedSerializationSchema((SerializationMetricWrapper)serializationSchema, partitionKeys), producerConfig, customPartitioner.orElse(null));
+        super(topicId, new CustomerKeyedSerializationSchema((SerializationMetricWrapper)serializationSchema, partitionKeys), producerConfig, customPartitioner.get());
         this.serializationMetricWrapper = (SerializationMetricWrapper) serializationSchema;
     }