feat: share kafka client on meta

e2e_test/source_legacy/basic/ddl.slt

@@ -30,7 +30,7 @@ Caused by these errors (recent errors listed first):
   1: gRPC request to meta service failed: Internal error
   2: failed to create source worker
   3: failed to parse json
-  4: missing field `properties.bootstrap.server`
+  4: missing field `topic`
 
 
 statement error

src/connector/Cargo.toml

@@ -71,7 +71,7 @@ jni = { version = "0.21.1", features = ["invocation"] }
 jsonbb = { workspace = true }
 jsonwebtoken = "9.2.0"
 maplit = "1.0.2"
-moka = { version = "0.12.0", features = ["future"] }
+moka = { version = "0.12.8", features = ["future"] }
 mongodb = { version = "2.8.2", features = ["tokio-runtime"] }
 mysql_async = { version = "0.34", default-features = false, features = [
     "default",

src/connector/src/connector_common/common.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::hash::{Hash, Hasher};
 use std::io::Write;
 use std::time::Duration;
 
@@ -61,7 +62,7 @@ use aws_types::SdkConfig;
 use risingwave_common::util::env_var::env_var_is_true;
 
 /// A flatten config map for aws auth.
-#[derive(Deserialize, Debug, Clone, WithOptions)]
+#[derive(Deserialize, Debug, Clone, WithOptions, PartialEq)]
 pub struct AwsAuthProps {
     #[serde(rename = "aws.region", alias = "region")]
     pub region: Option<String>,
@@ -161,21 +162,11 @@ impl AwsAuthProps {
 }
 
 #[serde_as]
-#[derive(Debug, Clone, Deserialize, WithOptions)]
-pub struct KafkaCommon {
+#[derive(Debug, Clone, Deserialize, WithOptions, PartialEq, Hash, Eq)]
+pub struct KafkaConnection {
     #[serde(rename = "properties.bootstrap.server", alias = "kafka.brokers")]
     pub brokers: String,
 
-    #[serde(rename = "topic", alias = "kafka.topic")]
-    pub topic: String,
-
-    #[serde(
-        rename = "properties.sync.call.timeout",
-        deserialize_with = "deserialize_duration_from_string",
-        default = "default_kafka_sync_call_timeout"
-    )]
-    pub sync_call_timeout: Duration,
-
     /// Security protocol used for RisingWave to communicate with Kafka brokers. Could be
     /// PLAINTEXT, SSL, SASL_PLAINTEXT or SASL_SSL.
     #[serde(rename = "properties.security.protocol")]
@@ -252,6 +243,20 @@ pub struct KafkaCommon {
 
 #[serde_as]
 #[derive(Debug, Clone, Deserialize, WithOptions)]
+pub struct KafkaCommon {
+    #[serde(rename = "topic", alias = "kafka.topic")]
+    pub topic: String,
+
+    #[serde(
+        rename = "properties.sync.call.timeout",
+        deserialize_with = "deserialize_duration_from_string",
+        default = "default_kafka_sync_call_timeout"
+    )]
+    pub sync_call_timeout: Duration,
+}
+
+#[serde_as]
+#[derive(Debug, Clone, Deserialize, WithOptions, PartialEq)]
 pub struct KafkaPrivateLinkCommon {
     /// This is generated from `private_link_targets` and `private_link_endpoint` in frontend, instead of given by users.
     #[serde(rename = "broker.rewrite.endpoints")]
@@ -269,7 +274,7 @@ pub struct RdKafkaPropertiesCommon {
     /// Maximum Kafka protocol request message size. Due to differing framing overhead between
     /// protocol versions the producer is unable to reliably enforce a strict max message limit at
     /// produce time and may exceed the maximum size by one message in protocol ProduceRequests,
-    /// the broker will enforce the the topic's max.message.bytes limit
+    /// the broker will enforce the topic's max.message.bytes limit
     #[serde(rename = "properties.message.max.bytes")]
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub message_max_bytes: Option<usize>,
@@ -316,7 +321,13 @@ impl RdKafkaPropertiesCommon {
     }
 }
 
-impl KafkaCommon {
+impl KafkaConnection {
+    pub fn get_hash(&self) -> u64 {
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        self.hash(&mut hasher);
+        hasher.finish()
+    }
+
     pub(crate) fn set_security_properties(&self, config: &mut ClientConfig) {
         // AWS_MSK_IAM
         if self.is_aws_msk_iam() {

src/connector/src/connector_common/mod.rs

@@ -19,9 +19,9 @@ pub use mqtt_common::{MqttCommon, QualityOfService as MqttQualityOfService};
 
 mod common;
 pub use common::{
-    AwsAuthProps, AwsPrivateLinkItem, KafkaCommon, KafkaPrivateLinkCommon, KinesisCommon,
-    MongodbCommon, NatsCommon, PulsarCommon, PulsarOauthCommon, RdKafkaPropertiesCommon,
-    PRIVATE_LINK_BROKER_REWRITE_MAP_KEY, PRIVATE_LINK_TARGETS_KEY,
+    AwsAuthProps, AwsPrivateLinkItem, KafkaCommon, KafkaConnection, KafkaPrivateLinkCommon,
+    KinesisCommon, MongodbCommon, NatsCommon, PulsarCommon, PulsarOauthCommon,
+    RdKafkaPropertiesCommon, PRIVATE_LINK_BROKER_REWRITE_MAP_KEY, PRIVATE_LINK_TARGETS_KEY,
 };
 
 mod iceberg;