feat(sink): support jdbc sink async truncate #13190
Conversation
|
| GitGuardian id | Secret | Commit | Filename | |
|---|---|---|---|---|
| 7648795 | Generic CLI Secret | 9b8ca15 | integration_tests/iceberg-cdc/run_test.sh | View secret |
| 7648795 | Generic CLI Secret | 9b8ca15 | integration_tests/iceberg-cdc/docker-compose.yml | View secret |
🛠 Guidelines to remediate hardcoded secrets
- Understand the implications of revoking this secret by investigating where it is used in your code.
- Replace and store your secrets safely. Learn here the best practices.
- Revoke and rotate these secrets.
- If possible, rewrite git history. Rewriting git history is not a trivial act. You might completely break other contributing developers' workflow and you risk accidentally deleting legitimate data.
To avoid such incidents in the future consider
- following these best practices for managing and storing secrets including API keys and other credentials
- install secret detection on pre-commit to catch secret before it leaves your machine and ease remediation.
🦉 GitGuardian detects secrets in your source code to help developers and security teams secure the modern development process. You are seeing this because you or someone else with access to this repository has authorized GitGuardian to scan your pull request.
Our GitHub checks need improvements? Share your feedbacks!
Codecov Report
@@ Coverage Diff @@
## main #13190 +/- ##
==========================================
- Coverage 68.00% 67.99% -0.02%
==========================================
Files 1521 1521
Lines 257858 257930 +72
==========================================
+ Hits 175364 175385 +21
- Misses 82494 82545 +51
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 6 files with indirect coverage changes 📣 Codecov offers a browser extension for seamless coverage viewing on GitHub. Try it in Chrome or Firefox today! |
...connector-node/connector-api/src/main/java/com/risingwave/connector/api/sink/SinkWriter.java
Show resolved
Hide resolved
| { ElasticSearch, ElasticSearchSink, "elasticsearch" } | ||
| { Cassandra, CassandraSink, "cassandra" } | ||
| { Jdbc, JdbcSink, "jdbc", |desc| { | ||
| desc.sink_type.is_append_only() |
There was a problem hiding this comment.
Are we planning to enable upsert jdbc sink decoupling as well?
There was a problem hiding this comment.
May keep the same as other sinks like kafka, kinesis for now. Current implementation does not support rewind on failure yet, and enabling it by default does not help resolve much of the current problems.
I think we can enable sink decouple for these sinks by default after rewind on failure is supported.
| )), | ||
| } => { | ||
| if let Some(metadata) = metadata { | ||
| warn!("get unexpected non-empty metadata: {:?}", metadata); |
There was a problem hiding this comment.
Is this really unexpected? I saw that java iceberg sink will return non-empty metadata.
There was a problem hiding this comment.
iceberg sink uses CoordinatedRemoteSinkWriter and will not be affected by this PR. Sink that use this RemoteLogSinker assume that they are not coordinated and should not return non-empty metadata.
There was a problem hiding this comment.
Rest lgtm. I think the related issue is #12060.
Btw I have a quick question about the failure isolation, what is the expected behavior if a sink fails to commit? For the case of JDBC sink fails due to incorrect data, I think we can avoid triggering a recovery with the help of log store. But does the sink will retry its work continuously?
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
After #13137
The return type of the
writemethod ofSinkWriterjava interface is changed tobool, wheretrueindicates that the batch is persisted in the external sink and is safe to truncate the log reader.A new item
BatchWrittenis added to the sink writer response, to indicate that the batch with the batch id has been persisted. The remote sink is refactored to poll the response and truncate asynchronously.Checklist
./risedev check(or alias,./risedev c)Documentation
Release note
If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.