feat: Connection for Kafka source & sink

[tabVersion]

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

[tabVersion]

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

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

[fuyufjh]

Overall LGTM

[fuyufjh]

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

[fuyufjh]

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.

[wcy-fdu]

Common part generally LGTM.

[wcy-fdu]

So we just reuse Source.connection_id?

[tabVersion]

Yes and same for the schema registry part and sink.