[fix-34344][kafka] remove partition key blank char.
@@ -39,6 +39,7 @@ import org.apache.flink.types.Row;
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 @@ public abstract class AbstractKafkaSink implements RetractStreamTableSink<Row>,
}
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 = keysStr.split(",");
+ String[] cleanedKeys = Arrays.stream(keys)
+ .map(x -> x.trim())
+ .toArray(String[]::new);
+ return cleanedKeys;
}
return null;
}
