feat: Connection for Kafka source & sink #19270

tabVersion · 2024-11-05T15:16:19Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

following #18975

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

introducing a new catalog CONNECTION. and have integrated with SECRET

please note that the legacy create connection to AWS PrivateLink has been deprecated in #18975.

new syntax

CREATE CONNECTION [ IF NOT EXISTS ] <connection name> with (
  type = 'kafka' / 'iceberg'
  some_atter_1 = secret <secret name>,
  some_attr_2 = '...'
);

planned support for Kafka, iceberg (by @chenzl25 ) and FS (@wcy-fdu ) at the first stage.

when creating source/sink from a connection, connector must match with the connection type & the ref key must be connection

create source s ( ... ) with (
  connector = 'kafka', # <- match with connection type
  connection = <connection name> # <- the key must be `profile` and use `connection` to mark ref to connection catalog. 
) format ... encode ...;

and the attrs defined in connection and source/table/sink cannot have overlap

create connection conn with ( type = 'kafka', a = 'a', b = 'b' );

# reject as overlap key `a` and `b`
create source s with ( connector = 'kafka', a = '1', 'b' = '2' , connection = conn ) format ... encode ... ;

one more thing
- we still allow creating source/table/sink without connection and the syntax remains unchanged.

to perform connection validate, we need a new kafka ACL: DESCRIBE CLUSTER (the privilege is auth via username and password, not related with consumer group. )

Connection stores KV in the catalog and validation only takes a copy.
When building source/sink/table, we first fill the KVs in connection catalog to the with options and then start the create procedure.

accepted kafka connection props

(connection related)

properties.bootstrap.server (only required)
properties.security.protocol
properties.ssl.endpoint.identification.algorithm
properties.ssl.ca.location
properties.ssl.ca.pem
properties.ssl.certificate.location
properties.ssl.certificate.pem
properties.ssl.key.location
properties.ssl.key.pem
properties.ssl.key.password
properties.sasl.mechanism
properties.sasl.username
properties.sasl.password
properties.sasl.kerberos.service.name
properties.sasl.kerberos.keytab
properties.sasl.kerberos.principal
properties.sasl.kerberos.kinit.cmd
properties.sasl.kerberos.min.time.before.relogin
properties.sasl.oauthbearer.config

(private link related)

privatelink.targets
privatelink.endpoint

handle_create_connection will do the private link resolve and remove both privatelink.targets and privatelink.endpoint and insert the broker.rewrite.endpoints to the props.

so if users specify privatelink.targets and privatelink.endpoint in connection, they cannot set it again when create source/table/sink.

(aws auth related: for msk)

aws.region
endpoint
aws.credentials.access_key_id
aws.credentials.secret_access_key
aws.credentials.session_token
aws.credentials.role.arn
aws.credentials.role.external_id

Signed-off-by: tabVersion <[email protected]>

Signed-off-by: tabversion <[email protected]>

Signed-off-by: tabVersion <[email protected]>

…ngwavelabs/risingwave into tab/share-kafka-client-enum

Signed-off-by: tabVersion <[email protected]>

…ngwavelabs/risingwave into tab/share-kafka-client-enum

e2e_test/source_inline/connection/ddl.slt

src/connector/src/connector_common/connection_util.rs

fuyufjh · 2024-11-19T09:29:43Z

src/connector/src/connector_common/connection_util.rs

+            config.create_with_context(client_ctx).await?;
+        if self.inner.is_aws_msk_iam() {
+            #[cfg(not(madsim))]
+            client.poll(Duration::from_secs(10)); // note: this is a blocking call


Perhaps the madsim needs to be updated?

src/frontend/src/catalog/connection_catalog.rs

src/connector/src/connector_common/connection_util.rs

fuyufjh · 2024-11-20T08:11:31Z

proto/catalog.proto

@@ -226,6 +236,19 @@ message Subscription {
  SubscriptionState subscription_state = 19;
 }

+message ConnectionParams {


This is kind of a weak-typed implementation, as the properties are kept as a map<string,string> here.

On the contrary, we may optionally make it strong-typed by defining connection's fields as a message (such as message KafkaConnection), and meanwhile, the corresponding struct (such as struct KafkaConnectionInner) would be substituted.

I tend to prefer the latter one, but I am not sure how to handle secrets. Comments are welcomed.

I had this idea before.
If we are going with a concrete type here, we have to accept a more complex impl when handling create source/sink/table, ie. we have to know an attr comes from either with-clause or connection catalog and make sure it won't collide. Besides, secret ref is a problem.

Current impl overcomes the prior problem by merging everything into hashmap and keeps the original path to create source/sink/table unchanged.

fuyufjh · 2024-11-20T08:14:45Z

proto/catalog.proto

@@ -246,6 +269,7 @@ message Connection {
  string name = 4;
  oneof info {


No need to keep the oneof now. May remove it to improve readability.

We are still using the proto in meta backup. Removing the oneof might break compatibility.

Well, then let's place connection_params outside the oneof at least

fuyufjh · 2024-11-20T08:22:31Z

src/meta/src/error.rs

@@ -132,6 +133,13 @@ pub enum MetaErrorInner {

    #[error("{0} has been deprecated, please use {1} instead.")]
    Deprecated(String, String),
+
+    #[error("Secret error: {0}")]
+    SecretError(


What's this for?

We will validate the connection on meta, which needs to fill in the secrets. So introducing Secret Error here for handling possible failure.

fuyufjh · 2024-11-20T08:27:38Z

proto/catalog.proto

@@ -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;


I think this should be added to SinkParam & SinkDesc instead, side-by-side with the Sink's properties.

Let me clear the logic for the related proto defs

Sink (proto/catalog.proto): def of sink in catalog for all sink, we will read props and resolve connection dep based on this message. And the message contains SinkFormatDesc. We already have connection.id field here.

We do solve connection ref when creating sink, so the props in Sink.properties contains the ones from connection catalog.

SinkFormatDesc (proto/catalog.proto): specifying format ... encode ... options (and contained by sink catalog). We ref a connection here for schema registry.

SinkDesc (proto/stream_plan.proto): It defines what field we dispatch to CNs. But we have already resolved the props from connection def in meta so there is no need to change things here.

SinkParam (proto/connector_service.proto): seems to define the transport protocol between kernel and connector node (most ref in Java). And ditto, the step is behind the meta node so no change is needed.

We do solve connection ref when creating sink, so the props in Sink.properties contains the ones from connection catalog.

I see. So here we expand all the connections into concrete parameters on creation, right? This is okay to me, but in this way, I think connection_id here in SinkFormatDesc should not appear as well, because all connection-related things are already eliminated on creation.

(It seems SinkFormatDesc::connection_id was indeed not used anywhere?)

Similar to the source side, the connector part and schema registry part will have independent connection, just like individual secret_refs.

we are offering syntax like

create source s(...) with ( connector = 'kafka', connection = <connector connection>, ... ) format ... encode ... ( connection = <schema registry connection / glue connection> )

Yeah, I know. What I mean is: should it be resolved and eliminated before converting into a proto message?

We do solve connection ref when creating sink, so the props in Sink.properties contains the ones from connection catalog.

Yeah, I know. What I mean is: should it be resolved and eliminated before converting into a proto message?

We resolve the props in the frontend and the catalog is persisted in the meta. We have to reserve a field to keep the connection_id to maintain the dependency. And to keep the design simple and align with the secret ref, I think adding connection.id for each connector and schema registry is the best practice.

& We are going to support alter connection and apply the changes to all related source/sink. Eliminating the connection.id in the catalog makes the step harder.

I see. You keep both connection_id and the resolved connection arguments in these message structures, right? It's acceptable to me but a bit counter-intuitive.

And to keep the design simple and align with the secret ref,

IIRC, secret ref doesn't keep the resolved plaintext secret at all. It always resolves whenever using a secret.

IIRC, secret ref doesn't keep the resolved plaintext secret at all. It always resolves whenever using a secret.

Oh, here is a some gap on resolve.
You can ref to resolve_connection_ref_and_secret_ref, resolve here means extracting xxx = secret <secret name> from "with props" to PbSecretRef (secret_id). We will do the same op for secrets coming from both connection catalog and with props.

fuyufjh · 2024-11-20T08:28:03Z

src/connector/src/sink/catalog/mod.rs

@@ -124,6 +124,7 @@ pub struct SinkFormatDesc {
    pub options: BTreeMap<String, String>,
    pub secret_refs: BTreeMap<String, PbSecretRef>,
    pub key_encode: Option<SinkEncode>,
+    pub connection_id: Option<u32>,


src/sqlparser/src/ast/value.rs

tabVersion · 2024-11-20T09:34:41Z

Please add or update some e2e test to cover this feature.

Does e2e_test/source_inline/connection/ddl.slt lack some scenes?

e2e_test/source_inline/connection/ddl.slt

Co-authored-by: tabversion <[email protected]>

Signed-off-by: tabversion <[email protected]>

fuyufjh

Overall LGTM

fuyufjh · 2024-11-26T05:50:52Z

proto/catalog.proto

@@ -246,6 +269,7 @@ message Connection {
  string name = 4;
  oneof info {


Well, then let's place connection_params outside the oneof at least

fuyufjh · 2024-11-26T05:56:39Z

proto/catalog.proto

@@ -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;


I see. You keep both connection_id and the resolved connection arguments in these message structures, right? It's acceptable to me but a bit counter-intuitive.

And to keep the design simple and align with the secret ref,

IIRC, secret ref doesn't keep the resolved plaintext secret at all. It always resolves whenever using a secret.

e2e_test/source_inline/connection/ddl.slt

This reverts commit dc3fd25.

wcy-fdu

Common part generally LGTM.

wcy-fdu · 2024-11-27T01:44:15Z

proto/catalog.proto

@@ -123,6 +126,8 @@ message Source {
    uint32 associated_table_id = 12;
  }
  string definition = 13;
+
+  // ref connection for connector
  optional uint32 connection_id = 14;


So we just reuse Source.connection_id?

Yes and same for the schema registry part and sink.

tabVersion and others added 30 commits October 22, 2024 21:37

share kafka client on meta

b9b2f79

Signed-off-by: tabVersion <[email protected]>

check kafka connection identical

ab46672

Signed-off-by: tabVersion <[email protected]>

fix

3373a58

Signed-off-by: tabVersion <[email protected]>

fix with props

c5203d6

Signed-off-by: tabVersion <[email protected]>

fix

3b6a6a2

Signed-off-by: tabVersion <[email protected]>

use connection hash as hashmap entry

3b3f725

Signed-off-by: tabVersion <[email protected]>

fix

968ed08

Signed-off-by: tabVersion <[email protected]>

fix

a41f3fc

Signed-off-by: tabversion <[email protected]>

rerun

6534ebf

Signed-off-by: tabVersion <[email protected]>

Merge branch 'tab/share-kafka-client-enum' of https://github.com/risi…

7249e78

…ngwavelabs/risingwave into tab/share-kafka-client-enum

Merge remote-tracking branch 'origin' into tab/share-kafka-client-enum

0024e1d

fix

ad8b989

Signed-off-by: tabVersion <[email protected]>

fix

ae1b70a

Signed-off-by: tabVersion <[email protected]>

better with options

58b5128

use kafka connection as hashkey

f115a0c

use moka

d128644

fix lint

ae9df41

fix

45295bc

fix

a9e34c7

Merge branch 'main' into tab/share-kafka-client-enum

d27ab90

Merge remote-tracking branch 'origin' into tab/share-kafka-client-enum

256485c

Merge branch 'tab/share-kafka-client-enum' of https://github.com/risi…

62cb953

…ngwavelabs/risingwave into tab/share-kafka-client-enum

remove get hash func

725e23c

migrate to Weak

832f66f

minor

73f0b7b

fix

ac1d63d

Merge remote-tracking branch 'origin' into tab/share-kafka-client-enum

35fb002

test bump quanta to 0.12.3

ec49096

update patch

16d8c42

moka 0.12.3

b3efda6

fuyufjh reviewed Nov 20, 2024

View reviewed changes

src/sqlparser/src/ast/value.rs Outdated Show resolved Hide resolved

This comment was marked as resolved.

Sign in to view

fuyufjh reviewed Nov 20, 2024

View reviewed changes

e2e_test/source_inline/connection/ddl.slt Outdated Show resolved Hide resolved

tabVersion and others added 4 commits November 20, 2024 22:43

resolve comments

795a79d

Merge remote-tracking branch 'origin' into tab/connection

c26c05b

Merge remote-tracking branch 'origin' into tab/connection

64a4e26

rename migration

28bb651

tabVersion mentioned this pull request Nov 25, 2024

refactor: make object ref outside ast::Value #19559

Merged

9 tasks

tabVersion requested a review from fuyufjh November 25, 2024 08:19

tabVersion and others added 3 commits November 25, 2024 17:09

refactor: make object ref outside ast::Value (#19559)

9c0e9df

Co-authored-by: tabversion <[email protected]>

Merge branch 'main' into tab/connection

e312ef2

rerun

d24fafb

Signed-off-by: tabversion <[email protected]>

fuyufjh approved these changes Nov 26, 2024

View reviewed changes

fuyufjh changed the title ~~feat: Connection for connector usage~~ feat: Connection for Kafka source & sink Nov 26, 2024

tabversion added 4 commits November 26, 2024 16:15

fix catalog proto

dc3fd25

Revert "fix catalog proto"

401660b

This reverts commit dc3fd25.

change syntax

08fb575

Merge remote-tracking branch 'origin' into tab/connection

48a28a2

wcy-fdu approved these changes Nov 27, 2024

View reviewed changes

tabversion added 2 commits November 28, 2024 17:37

Merge remote-tracking branch 'origin' into tab/connection

d5be997

test

2f0a4a8

tabVersion added this pull request to the merge queue Nov 28, 2024

Merged via the queue into main with commit 01b703d Nov 28, 2024
33 of 34 checks passed

tabVersion deleted the tab/connection branch November 28, 2024 14:17

neverchanje mentioned this pull request Nov 28, 2024

Document: feat: Connection for Kafka source & sink risingwavelabs/risingwave-docs#100

Open

This was referenced Dec 2, 2024

tracking: Connection #19637

Open

feat: telemetry for connection create and ref #19638

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Connection for Kafka source & sink #19270

feat: Connection for Kafka source & sink #19270

tabVersion commented Nov 5, 2024 •

edited

Loading

fuyufjh Nov 19, 2024

fuyufjh Nov 20, 2024

tabVersion Nov 20, 2024

fuyufjh Nov 20, 2024

tabVersion Nov 25, 2024

fuyufjh Nov 26, 2024

fuyufjh Nov 20, 2024

tabVersion Nov 20, 2024

fuyufjh Nov 20, 2024

tabVersion Nov 20, 2024

fuyufjh Nov 20, 2024 •

edited

Loading

tabVersion Nov 20, 2024

fuyufjh Nov 21, 2024

tabVersion Nov 21, 2024 •

edited

Loading

fuyufjh Nov 26, 2024

tabVersion Nov 26, 2024

fuyufjh Nov 20, 2024

This comment was marked as resolved.

tabVersion commented Nov 20, 2024

fuyufjh left a comment

fuyufjh Nov 26, 2024

fuyufjh Nov 26, 2024

wcy-fdu left a comment

wcy-fdu Nov 27, 2024

tabVersion Nov 27, 2024

		@@ -246,6 +269,7 @@ message Connection {
		string name = 4;
		oneof info {

feat: Connection for Kafka source & sink #19270

feat: Connection for Kafka source & sink #19270

Conversation

tabVersion commented Nov 5, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fuyufjh Nov 20, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion Nov 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

This comment was marked as resolved.

tabVersion commented Nov 20, 2024

fuyufjh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcy-fdu left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tabVersion commented Nov 5, 2024 •

edited

Loading

fuyufjh Nov 20, 2024 •

edited

Loading

tabVersion Nov 21, 2024 •

edited

Loading