last commit

Author: dapeng

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

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

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

@@ -28,7 +28,6 @@ public byte[] serializeKey(Row element) {
     }
         SerializationSchema<Row> serializationSchema = serializationMetricWrapper.getSerializationSchema();
         if(serializationSchema instanceof JsonRowSerializationSchema){
-            System.out.println("serial start, element = " + element);
             return serializeJsonKey((JsonRowSerializationSchema) serializationSchema, element);
         }
         return null;
@@ -46,11 +45,9 @@ private byte[] serializeJsonKey(JsonRowSerializationSchema jsonRowSerializationS
         try {
             byte[] data = jsonRowSerializationSchema.serialize(element);
             ObjectNode objectNode = mapper.readValue(data, ObjectNode.class);
-            System.out.println("objectNode = " + objectNode);
             StringBuilder sb = new StringBuilder();
             for(String key : partitionKeys){
                 if(objectNode.has(key)){
-                    System.out.println("key = " + key+ ", value = " + objectNode.get(key));
                     sb.append(objectNode.get(key));
                 }
             }

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

@@ -82,7 +82,7 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
             properties.setProperty(key, kafka10SinkTableInfo.getKafkaParam(key));
         }
 
-        this.partitioner = Optional.of(getFlinkPartitioner(kafka10SinkTableInfo));
+        this.partitioner = Optional.of(new CustomerFlinkPartition<>());
         this.partitionKeys = getPartitionKeys(kafka10SinkTableInfo);
         this.fieldNames = kafka10SinkTableInfo.getFields();
         TypeInformation[] types = new TypeInformation[kafka10SinkTableInfo.getFields().length];
@@ -142,17 +142,7 @@ public TableSink<Tuple2<Boolean, Row>> configure(String[] fieldNames, TypeInform
         this.fieldTypes = fieldTypes;
         return this;
     }
-
-    private FlinkKafkaPartitioner getFlinkPartitioner(KafkaSinkTableInfo kafkaSinkTableInfo){
-        System.out.println("enablePartition =" + kafkaSinkTableInfo.getEnableKeyPartition());
-        if("true".equalsIgnoreCase(kafkaSinkTableInfo.getEnableKeyPartition())){
-            return new CustomerFlinkPartition<>();
-        }
-        return new FlinkFixedPartitioner<>();
-    }
-
     private String[] getPartitionKeys(KafkaSinkTableInfo kafkaSinkTableInfo){
-        System.out.println("partitionKeys =" + kafkaSinkTableInfo.getPartitionKeys());
         if(StringUtils.isNotBlank(kafkaSinkTableInfo.getPartitionKeys())){
             return StringUtils.split(kafkaSinkTableInfo.getPartitionKeys(), ',');
         }