Merge branch 'hotfix_1.10_4.0_34344' into 1.10_release_4.0.x

Author: gituser

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

@@ -39,6 +39,7 @@
 import org.apache.flink.util.Preconditions;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Optional;
 import java.util.Properties;
@@ -113,8 +114,13 @@ protected TableSchema buildTableSchema(String[] fieldNames, TypeInformation<?>[]
     }
 
     protected String[] getPartitionKeys(KafkaSinkTableInfo kafkaSinkTableInfo) {
-        if (StringUtils.isNotBlank(kafkaSinkTableInfo.getPartitionKeys())) {
-            return StringUtils.split(kafkaSinkTableInfo.getPartitionKeys(), ',');
+        String keysStr = kafkaSinkTableInfo.getPartitionKeys();
+        if (StringUtils.isNotBlank(keysStr)) {
+            String[] keys = StringUtils.split(keysStr, ",");
+            String[] cleanedKeys = Arrays.stream(keys)
+                .map(x -> x.trim())
+                .toArray(String[]::new);
+            return cleanedKeys;
         }
         return null;
     }