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

What's changed and what's your intention?

• Support new syntax to create a Kafka source with the private link.

CREATE CONNECTION my_connection with (
    type = 'privatelink',
    provider = 'aws',
    service.name = 'xxx'
);

CREATE SOURCE tcp_metrics_rw (
    device_id VARCHAR,
    metric_name VARCHAR,
    report_time TIMESTAMP,
    metric_value DOUBLE PRECISION
) WITH (
  connector = 'kafka',
  topic = 'tcp_metrics',
  properties.bootstrap.server = 'ip1:9092,ip2:9092',
  connection.name = 'my_connection',
  privatelink.targets = '[{"port": 8001}, {"port": 8002}]',
  scan.startup.mode = 'earliest'
) ROW FORMAT JSON;

• And since users have to enable the cross-zone load balancing to allow private link works in our cloud, the availability zone arg in the WITH clause is useless, just remove it.
• The drop source procedure will decrease the refcnt of the connection. feat(source): create source with connection/drop connection #9128

#8771

Checklist For Contributors

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• I have demonstrated that backward compatibility is not broken by breaking changes and created issues to track deprecated features to be removed in the future. (Please refer to the issue)
• All checks passed in ./risedev check (or alias, ./risedev c)

Checklist For Reviewers

• I have requested macro/micro-benchmarks as this PR can affect performance substantially, and the results are shown.

Documentation