[INLONG-11576][Sort] Support KV separator for kafka sink

vernedeng · Dec 5, 2024 · 361bef3 · 361bef3
1 parent 7ccbd97
commit 361bef3
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/...lone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java b/...lone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
@@ -21,6 +21,7 @@
 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 static KafkaIdConfig create(DataFlowConfig dataFlowConfig) {
         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()