fix(mysql-cdc): use connection pool to avoid data loss caused by unsafe cancellation #22128
+142
−25
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
CI failed, under investigation |
…bs/risingwave into wcy/mysql_async_fix
hzxa21
reviewed
Jun 10, 2025
| @@ -375,10 +374,15 @@ impl ExternalTableReader for MySqlExternalTableReader { | |||
| ) -> BoxStream<'_, ConnectorResult<OwnedRow>> { | |||
| self.snapshot_read_inner(table_name, start_pk, primary_keys, limit) | |||
| } | |||
|
|
|||
| async fn disconnect(&self) -> ConnectorResult<()> { | |||
There was a problem hiding this comment.
You can pass ownership of self here so that you don't need to clone the pool: async fn disconnect(self)
Comment on lines
+247
to
+250
| async fn disconnect(&self) -> StreamExecutorResult<()> { | ||
| self.reader.disconnect().await?; | ||
| Ok(()) | ||
| } |
| @@ -200,6 +200,12 @@ pub type CdcOffsetParseFunc = Box<dyn Fn(&str) -> ConnectorResult<CdcOffset> + S | |||
| pub trait ExternalTableReader { | |||
| async fn current_cdc_offset(&self) -> ConnectorResult<CdcOffset>; | |||
|
|
|||
| // Currently, MySQL cdc uses a connection pool to manage connections to MySQL, and other CDC processes do not require the disconnect step for now. | |||
| #[allow(clippy::unused_async)] | |||
There was a problem hiding this comment.
Why do we need this? I thought it is always used.
There was a problem hiding this comment.
This is because other cdc(pg, mssql..) does not need this method.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
In the implementation of MySQL CDC, we use mysql_async to read historical data from MySQL. We utilize the exec_drop interface, but there is a cancellation safety issue: if a drop occurs while exec_drop is running, and if the connection is not rebuilt, subsequent calls to read_stream will return empty. At this point, the upper layer will mistakenly believe that the backfill has ended, leading to data loss.
One way to simulate and reproduce this error is to construct a case where exec_drop is dropped:
In the above example, after exec_drop times out and is dropped, using this connection afterwards will result in an error:
Io(Io(Custom { kind: UnexpectedEof, error: "can't parse: buf doesn't have enough data" }))After reviewing the official documentation for mysql_async, I believe we should use the recommended connection pool. When we need a connection, we can use
get_connand then drop the connection withdrop(conn)once we're done. Additionally, after backfilling and reading all historical data, we should first runpool.disconnectbefore dropping snapshot_stream. This approach can avoid cancellation safety issues and elegantly reclaim resources.Note that, it's a bit hard to construct a scenario where a drop occurs exactly during the execution of
exec_drop, so we constructed a unit test using minimal reproducible code and ran it using the MySQL environment ine2e-source-test.Checklist
Documentation
Release note