optimize kafka connector

Xiong-Da · Xiong-Da · commit 807bcdbc0aff · 2026-04-11T17:11:06.000+08:00
diff --git a/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/calcite/KafkaCalciteTable.java b/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/calcite/KafkaCalciteTable.java
@@ -81,9 +81,9 @@ private static synchronized KafkaProducer<String, String> openKafkaProducer(Kafk
 
         Properties props = new Properties();
         props.put("bootstrap.servers", kafkaConfig.bootstrapServers);
-        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
-        props.put("linger.ms", 5000);   //todo read table config
+        props.put("key.serializer", kafkaConfig.keySerializer);
+        props.put("value.serializer", kafkaConfig.valueSerializer);
+        props.put("linger.ms", kafkaConfig.lingerMs);
 
         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
         kafkaProducerMap.put(kafkaConfig.bootstrapServers, producer);
diff --git a/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/config/KafkaConfig.java b/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/config/KafkaConfig.java
@@ -9,4 +9,7 @@ public class KafkaConfig {
     public String topic;
     public String format;
     public List<FieldSchema> fieldSchemas;
+    public String keySerializer;
+    public String valueSerializer;
+    public int lingerMs;
 }
diff --git a/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/config/KafkaOptions.java b/sqlrec-connectors/sqlrec-connector-kafka/src/main/java/com/sqlrec/connectors/kafka/config/KafkaOptions.java
@@ -28,13 +28,37 @@ public class KafkaOptions {
             null,
             String.class
     );
+    public static final ConfigOption<String> KEY_SERIALIZER = new ConfigOption<>(
+            "properties.producer.key.serializer",
+            "org.apache.kafka.common.serialization.StringSerializer",
+            "Kafka producer key serializer",
+            null,
+            String.class
+    );
+    public static final ConfigOption<String> VALUE_SERIALIZER = new ConfigOption<>(
+            "properties.producer.value.serializer",
+            "org.apache.kafka.common.serialization.StringSerializer",
+            "Kafka producer value serializer",
+            null,
+            String.class
+    );
+    public static final ConfigOption<Integer> LINGER_MS = new ConfigOption<>(
+            "properties.producer.linger.ms",
+            5000,
+            "Kafka producer linger ms",
+            null,
+            Integer.class
+    );
 
 
     public static KafkaConfig getKafkaConfig(Map<String, String> options) {
         KafkaConfig kafkaConfig = new KafkaConfig();
         kafkaConfig.bootstrapServers = BOOTSTRAP_SERVERS.getValue(options);
         kafkaConfig.topic = TOPIC.getValue(options);
         kafkaConfig.format = FORMAT.getValue(options);
+        kafkaConfig.keySerializer = KEY_SERIALIZER.getValue(options);
+        kafkaConfig.valueSerializer = VALUE_SERIALIZER.getValue(options);
+        kafkaConfig.lingerMs = LINGER_MS.getValue(options);
 
         return kafkaConfig;
     }

Original file line number	Diff line number	Diff line change
`@@ -9,4 +9,7 @@ public class KafkaConfig {`
`9`	`9`	`public String topic;`
`10`	`10`	`public String format;`
`11`	`11`	`public List<FieldSchema> fieldSchemas;`
	`12`	`+ public String keySerializer;`
	`13`	`+ public String valueSerializer;`
	`14`	`+ public int lingerMs;`
`12`	`15`	`}`