Merge branch 'feat_1.8_keyPartition_v2' into 'feat_keyPartition_mergeTest'

fix 缩进和日志



See merge request !258

Author: dapeng

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

@@ -23,10 +23,8 @@ public void open(int parallelInstanceId, int parallelInstances) {
     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.");
         if(key == null){
-            System.out.println("partitionKey=null");
             return partitions[this.parallelInstanceId % partitions.length];
         }
-        System.out.println("partitionKey=" + new String(key));
         return partitions[Math.abs(new String(key).hashCode()) % partitions.length];
     }
 

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

@@ -8,9 +8,13 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
 import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class CustomerKeyedSerializationSchema implements KeyedSerializationSchema<Row> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(CustomerKeyedSerializationSchema.class);
+
     private static final long serialVersionUID = 1L;
     private final SerializationMetricWrapper serializationMetricWrapper;
     private String[] partitionKeys;
@@ -52,10 +56,9 @@ private byte[] serializeJsonKey(JsonRowSerializationSchema jsonRowSerializationS
                 }
             }
             return sb.toString().getBytes();
-        }catch (Exception e){
-
+        } catch (Exception e){
+            LOG.error("serializeJsonKey error", e);
         }
         return null;
-
     }
 }