fix

Author: luoluoyuyu

src/sql/connector/sink/kafka.rs

@@ -140,6 +140,7 @@ impl SinkProvider for KafkaSinkConnector {
         client_configs.remove(opt::CHECKPOINT_INTERVAL_MS);
         client_configs.remove(opt::PIPELINE_PARALLELISM);
         client_configs.remove(opt::KEY_BY_PARALLELISM);
+        client_configs.remove(opt::FORMAT);
 
         Ok(ConnectorConfig::KafkaSink(KafkaSinkConfig {
             topic,

src/sql/connector/source/kafka.rs

@@ -163,6 +163,7 @@ impl SourceProvider for KafkaSourceConnector {
         client_configs.remove(opt::CHECKPOINT_INTERVAL_MS);
         client_configs.remove(opt::PIPELINE_PARALLELISM);
         client_configs.remove(opt::KEY_BY_PARALLELISM);
+        client_configs.remove(opt::FORMAT);
 
         Ok(ConnectorConfig::KafkaSource(KafkaSourceConfig {
             topic,

tests/integration/test/wasm/python_sdk/conftest.py

@@ -92,9 +92,19 @@ def __init__(self, client: FsClient):
         self._client = client
         self._registered: Set[str] = set()
 
+    def __contains__(self, name: str) -> bool:
+        return name in self._registered
+
     def append(self, name: str) -> None:
         self._registered.add(name)
 
+    def extend(self, names) -> None:
+        for name in names:
+            self._registered.add(name)
+
+    def remove(self, name: str) -> None:
+        self._registered.discard(name)
+
     def register(self, name: str) -> None:
         self._registered.add(name)