feat: Connection for Kafka source & sink

Cargo.toml

@@ -155,7 +155,11 @@ arrow-udf-flight = "0.4"
 clap = { version = "4", features = ["cargo", "derive", "env"] }
 # Use a forked version which removes the dependencies on dynamo db to reduce
 # compile time and binary size.
-deltalake = { version = "0.20.1", features = ["s3", "gcs", "datafusion"] }
+deltalake = { version = "0.20.1", features = [
+    "s3",
+    "gcs",
+    "datafusion",
+] }
 itertools = "0.13.0"
 jsonbb = "0.1.4"
 lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "2682b85" }

e2e_test/source_inline/connection/ddl.slt

@@ -0,0 +1,114 @@
+control substitution on
+
+# for non-shared source
+statement ok
+set streaming_use_shared_source to false;
+
+statement ok
+create secret sec_broker with (backend = 'meta') as '${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}';
+
+statement error unknown field `foo`
+create connection conn with (type = 'kafka', properties.bootstrap.server = secret sec_broker, foo = 'bar');
+
+statement error Connection type "kinesis" is not supported
+create connection conn with (type = 'kinesis');
+
+statement ok
+create connection conn with (type = 'kafka', properties.bootstrap.server = secret sec_broker, properties.security.protocol = 'plaintext');
+
+query TTT
+select "name", "type_" from rw_catalog.rw_connections;
+----
+conn CONNECTION_TYPE_KAFKA
+
+# unstable test serialization due to iter on hashmap
+# the "connectiond_params" looks like:
+# {"properties.bootstrap.server":"SECRET sec_broker AS TEXT","properties.security.protocol":"plaintext"}
+
+statement error Permission denied: PermissionDenied: secret used by 1 other objects.
+drop secret sec_broker;
+
+statement error expect 'profile' as the key for connection ref, but got: connection
+create table t1 (a int, b varchar) with (
+  connector = 'kafka',
+  connection = connection conn,
+  topic = 'connection_ddl_1')
+format plain encode json;
+
+statement error Duplicated key in both WITH clause and Connection catalog: properties.security.protocol
+create table t1 (a int, b varchar) with (
+  connector = 'kafka',
+  profile = connection conn,
+  topic = 'connection_ddl_1',
+  properties.security.protocol = 'plaintext')
+format plain encode json;
+
+statement error connector kinesis and connection type Kafka are not compatible
+create table t1 (a int, b varchar) with (
+  connector = 'kinesis',
+  profile = connection conn,
+  stream = 'connection_ddl_1',
+  region = 'us-east-1')
+format plain encode json;
+
+system ok
+rpk topic create connection_ddl_1 -p 1
+
+statement ok
+create table t1 (a int, b varchar) with (
+  connector = 'kafka',
+  profile = connection conn,
+  topic = 'connection_ddl_1')
+format plain encode json;
+
+statement error Permission denied: PermissionDenied: connection used by 1 other objects.
+drop connection conn;
+
+# Connection & Source & Sink will have independent rely on the secret
+statement error Permission denied: PermissionDenied: secret used by 2 other objects.
+drop secret sec_broker;
+
+statement ok
+create table data_table (a int, b varchar);
+
+statement ok
+insert into data_table values (1, 'a'), (2, 'b'), (3, 'c');
+
+statement ok
+flush;
+
+statement ok
+create sink sink_kafka from data_table with (
+  connector = 'kafka',
+  profile = connection conn,
+  topic = 'connection_ddl_1'
+) format plain encode json (
+  force_append_only='true'
+);
+
+sleep 3s
+
+query IT rowsort
+select a, b from t1;
+----
+1 a
+2 b
+3 c
+
+statement ok
+drop sink sink_kafka
+
+statement ok
+drop table data_table;
+
+statement ok
+drop table t1;
+
+statement ok
+drop connection conn;
+
+statement ok
+drop secret sec_broker;
+
+statement ok
+set streaming_use_shared_source to true;

proto/catalog.proto

@@ -92,6 +92,9 @@ message StreamSourceInfo {
   // Handle the source relies on any sceret. The key is the propertity name and the value is the secret id and type.
   // For format and encode options.
   map<string, secret.SecretRef> format_encode_secret_refs = 16;
+
+  // ref connection for schema registry
+  optional uint32 connection_id = 17;
 }
 
 message Source {
@@ -123,6 +126,8 @@ message Source {
     uint32 associated_table_id = 12;
   }
   string definition = 13;
+
+  // ref connection for connector
   optional uint32 connection_id = 14;
 
   optional uint64 initialized_at_epoch = 15;
@@ -156,6 +161,9 @@ message SinkFormatDesc {
   optional plan_common.EncodeType key_encode = 4;
   // Secret used for format encode options.
   map<string, secret.SecretRef> secret_refs = 5;
+
+  // ref connection for schema registry
+  optional uint32 connection_id = 6;
 }
 
 // the catalog of the sink. There are two kind of schema here. The full schema is all columns
@@ -178,6 +186,8 @@ message Sink {
   uint32 owner = 11;
   map<string, string> properties = 12;
   string definition = 13;
+
+  // ref connection for connector
   optional uint32 connection_id = 14;
   optional uint64 initialized_at_epoch = 15;
   optional uint64 created_at_epoch = 16;
@@ -226,6 +236,19 @@ message Subscription {
   SubscriptionState subscription_state = 19;
 }
 
+message ConnectionParams {
+  enum ConnectionType {
+    CONNECTION_TYPE_UNSPECIFIED = 0;
+    CONNECTION_TYPE_KAFKA = 1;
+    CONNECTION_TYPE_ICEBERG = 2;
+    CONNECTION_TYPE_SCHEMA_REGISTRY = 3;
+  }
+
+  ConnectionType connection_type = 1;
+  map<string, string> properties = 2;
+  map<string, secret.SecretRef> secret_refs = 3;
+}
+
 message Connection {
   message PrivateLinkService {
     enum PrivateLinkProvider {
@@ -246,6 +269,7 @@ message Connection {
   string name = 4;
   oneof info {
     PrivateLinkService private_link_service = 5 [deprecated = true];
+    ConnectionParams connection_params = 7;
   }
   uint32 owner = 6;
 }

proto/ddl_service.proto

@@ -429,7 +429,8 @@ message CreateConnectionRequest {
   uint32 database_id = 2;
   uint32 schema_id = 3;
   oneof payload {
-    PrivateLink private_link = 4;
+    PrivateLink private_link = 4 [deprecated = true];
+    catalog.ConnectionParams connection_params = 6;
   }
   uint32 owner_id = 5;
 }

src/bench/sink_bench/main.rs

@@ -487,6 +487,7 @@ fn mock_from_legacy_type(
             options: Default::default(),
             secret_refs: Default::default(),
             key_encode: None,
+            connection_id: None,
         }))
     } else {
         SinkFormatDesc::from_legacy_type(connector, r#type)

src/connector/src/connector_common/common.rs

@@ -163,7 +163,7 @@ impl AwsAuthProps {
 
 #[serde_as]
 #[derive(Debug, Clone, Deserialize, WithOptions, PartialEq, Hash, Eq)]
-pub struct KafkaConnection {
+pub struct KafkaConnectionProps {
     #[serde(rename = "properties.bootstrap.server", alias = "kafka.brokers")]
     pub brokers: String,
 
@@ -244,6 +244,7 @@ pub struct KafkaConnection {
 #[serde_as]
 #[derive(Debug, Clone, Deserialize, WithOptions)]
 pub struct KafkaCommon {
+    // connection related props are moved to `KafkaConnection`
     #[serde(rename = "topic", alias = "kafka.topic")]
     pub topic: String,
 
@@ -256,7 +257,7 @@ pub struct KafkaCommon {
 }
 
 #[serde_as]
-#[derive(Debug, Clone, Deserialize, WithOptions, PartialEq)]
+#[derive(Debug, Clone, Deserialize, WithOptions, PartialEq, Hash, Eq)]
 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")]
@@ -321,7 +322,32 @@ impl RdKafkaPropertiesCommon {
     }
 }
 
-impl KafkaConnection {
+impl KafkaConnectionProps {
+    #[cfg(test)]
+    pub fn test_default() -> Self {
+        Self {
+            brokers: "localhost:9092".to_string(),
+            security_protocol: None,
+            ssl_ca_location: None,
+            ssl_certificate_location: None,
+            ssl_key_location: None,
+            ssl_ca_pem: None,
+            ssl_certificate_pem: None,
+            ssl_key_pem: None,
+            ssl_key_password: None,
+            ssl_endpoint_identification_algorithm: None,
+            sasl_mechanism: None,
+            sasl_username: None,
+            sasl_password: None,
+            sasl_kerberos_service_name: None,
+            sasl_kerberos_keytab: None,
+            sasl_kerberos_principal: None,
+            sasl_kerberos_kinit_cmd: None,
+            sasl_kerberos_min_time_before_relogin: None,
+            sasl_oathbearer_config: None,
+        }
+    }
+
     pub(crate) fn set_security_properties(&self, config: &mut ClientConfig) {
         // AWS_MSK_IAM
         if self.is_aws_msk_iam() {