optimize kafka connector

Xiong-Da · Xiong-Da · commit 6dc0c1a1490c · 2026-04-11T17:29:08.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
@@ -67,16 +67,22 @@ public Expression getExpression(SchemaPlus schema, String tableName, Class clazz
                 tableName, clazz);
     }
 
+    private static String getProducerConfigKey(KafkaConfig config) {
+        return config.bootstrapServers + "|" + config.keySerializer + "|" + config.valueSerializer + "|" + config.lingerMs;
+    }
+
     public static KafkaProducer<String, String> getKafkaProducer(KafkaConfig kafkaConfig) {
-        if (kafkaProducerMap.containsKey(kafkaConfig.bootstrapServers)) {
-            return kafkaProducerMap.get(kafkaConfig.bootstrapServers);
+        String configKey = getProducerConfigKey(kafkaConfig);
+        if (kafkaProducerMap.containsKey(configKey)) {
+            return kafkaProducerMap.get(configKey);
         }
         return openKafkaProducer(kafkaConfig);
     }
 
     private static synchronized KafkaProducer<String, String> openKafkaProducer(KafkaConfig kafkaConfig) {
-        if (kafkaProducerMap.containsKey(kafkaConfig.bootstrapServers)) {
-            return kafkaProducerMap.get(kafkaConfig.bootstrapServers);
+        String configKey = getProducerConfigKey(kafkaConfig);
+        if (kafkaProducerMap.containsKey(configKey)) {
+            return kafkaProducerMap.get(configKey);
         }
 
         Properties props = new Properties();
@@ -86,7 +92,7 @@ private static synchronized KafkaProducer<String, String> openKafkaProducer(Kafk
         props.put("linger.ms", kafkaConfig.lingerMs);
 
         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
-        kafkaProducerMap.put(kafkaConfig.bootstrapServers, producer);
+        kafkaProducerMap.put(configKey, producer);
         return producer;
     }
 
