feat(source): Use a private link connection to create Kafka source #9119
Conversation
|
LGTM :) |
| @@ -688,45 +689,57 @@ where | |||
| .get(kafka_props_broker_key(properties)) | |||
| .cloned() | |||
| .ok_or(MetaError::from(anyhow!( | |||
| "Must specify brokers in WITH clause", | |||
| "Must specify brokers property in WITH clause", | |||
There was a problem hiding this comment.
| "Must specify brokers property in WITH clause", | |
| "Must specify \"properties.bootstrap.server\" in WITH clause", |
By the way, why is there a properties.? Looks redundant..
There was a problem hiding this comment.
Fixed. Hmm, I have no idea. It has been there for a long time. cc @tabVersion
There was a problem hiding this comment.
it is following Flink's naming style
Codecov Report
@@ Coverage Diff @@
## main #9119 +/- ##
==========================================
- Coverage 70.76% 70.73% -0.04%
==========================================
Files 1195 1195
Lines 197607 197673 +66
==========================================
- Hits 139834 139818 -16
- Misses 57773 57855 +82
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 6 files with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
Why do we use json format here if only port is provided? This seems inconsistent with the format of other properties. Can we just provide something like "8001, 8002" here. |
The cloud team doesn't have a final conclusion about the availability zone issue. Use a Json here reserves the flexibility to add more args for the target in the future. cc @mikechesterwang |
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
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;availability zonearg in the WITH clause is useless, just remove it.#8771
Checklist For Contributo 8000 rs
./risedev check(or alias,./risedev c)Checklist For Reviewers
Documentation
Click here for Documentation
Types of user-facing changes
Please keep the types that apply to your changes, and remove the others.
Release note
CREATE CONNECTIONthat allow users to create a privatelink connection in the cloudCREATE SOURCESQL command