test system.out

Author: dapeng

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

@@ -23,12 +23,12 @@ public CustomerKeyedSerializationSchema(SerializationMetricWrapper serialization
     }
 
     public byte[] serializeKey(Row element) {
-        System.out.println("element = " + element+"|partitionKeys=" + partitionKeys);
         if(partitionKeys == null || partitionKeys.length <=0){
             return null;
-        }
+    }
         SerializationSchema<Row> serializationSchema = serializationMetricWrapper.getSerializationSchema();
         if(serializationSchema instanceof JsonRowSerializationSchema){
+            System.out.println("serial start, element = " + element);
             return serializeJsonKey((JsonRowSerializationSchema) serializationSchema, element);
         }
         return null;
@@ -46,13 +46,15 @@ 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));
                 }
             }
-            return null;
+            return sb.toString().getBytes();
         }catch (Exception e){
 
         }

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

@@ -102,7 +102,6 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
         if (parallelism != null) {
             this.parallelism = parallelism;
         }
-
         this.kafkaProducer010 = (FlinkKafkaProducer010<Row>) new KafkaProducer010Factory().createKafkaProducer(kafka10SinkTableInfo, getOutputType().getTypeAt(1), properties, partitioner, partitionKeys);
         return this;
     }
@@ -145,13 +144,15 @@ public TableSink<Tuple2<Boolean, Row>> configure(String[] fieldNames, TypeInform
     }
 
     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(), ',');
         }