fix blank and hashcode question

Author: dapeng

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

@@ -22,10 +22,11 @@ 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.");
+        System.out.println("partitionKey=" + new String(key));
         if(key == null){
             return partitions[this.parallelInstanceId % partitions.length];
         }
-        return partitions[key.hashCode() % partitions.length];
+        return partitions[new String(key).hashCode() % partitions.length];
     }
 
     public boolean equals(Object o) {

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

@@ -48,7 +48,7 @@ private byte[] serializeJsonKey(JsonRowSerializationSchema jsonRowSerializationS
             StringBuilder sb = new StringBuilder();
             for(String key : partitionKeys){
                 if(objectNode.has(key)){
-                    sb.append(objectNode.get(key));
+                    sb.append(objectNode.get(key.trim()));
                 }
             }
             return sb.toString().getBytes();