(inlong) branch master updated: [INLONG-11576][Sort] Support KV separator for kafka sink (#11579)

[aloyszhang]

This is an automated email from the ASF dual-hosted git repository.

aloyszhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new 0da6faece7 [INLONG-11576][Sort] Support KV separator for kafka sink 
(#11579)
0da6faece7 is described below

commit 0da6faece7fa7ee1d15929a8083ad0a6d92c9512
Author: vernedeng <verned...@apache.org>
AuthorDate: Thu Dec 5 21:57:47 2024 +0800

    [INLONG-11576][Sort] Support KV separator for kafka sink (#11579)
---
 .../org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java    | 3 +++
 1 file changed, 3 insertions(+)

diff --git 
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
 
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
index a1c01bdb73..c6d29a9c6d 100644
--- 
a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
+++ 
b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
@@ -21,6 +21,7 @@ import org.apache.inlong.common.enums.DataTypeEnum;
 import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.dataType.CsvConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.dataType.DataTypeConfig;
+import org.apache.inlong.common.pojo.sort.dataflow.dataType.KvConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.KafkaSinkConfig;
 import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
@@ -66,6 +67,8 @@ public class KafkaIdConfig extends IdConfig {
         String separator = DEFAULT_SEPARATOR;
         if (dataTypeConfig instanceof CsvConfig) {
             separator = String.valueOf(((CsvConfig) 
dataTypeConfig).getDelimiter());
+        } else if (dataTypeConfig instanceof KvConfig) {
+            separator = String.valueOf(((KvConfig) 
dataTypeConfig).getEntrySplitter());
         }
 
         return KafkaIdConfig.builder()