risingwave_connector/source/cdc/
mysql.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeMap;
16
17use rand::Rng;
18use risingwave_pb::secret::PbSecretRef;
19
20use crate::error::ConnectorResult;
21use crate::source::cdc::{
22    CdcSourceDdlValidator, CdcSourceRuntimeFlags, CdcSourceTypeTrait, Mysql,
23    check_alter_with_options_inner, check_common_with_options,
24};
25
26// ------- Runtime flags -------
27
28impl CdcSourceRuntimeFlags for Mysql {
29    const CAN_BACKFILL: bool = true;
30    const CAN_USE_TRANSACTION_METADATA: bool = true;
31    const NEEDS_NON_EMPTY_INITIAL_OFFSET: bool = true;
32}
33
34// ------- DDL validation -------
35
36impl CdcSourceDdlValidator for Mysql {
37    fn check_and_fill_create_with_options(
38        props: &mut BTreeMap<String, String>,
39    ) -> ConnectorResult<()> {
40        check_common_with_options(props)?;
41        // Generate a random server id for mysql cdc source if needed
42        // `server.id` (in the range from 1 to 2^32 - 1). This value MUST be unique across whole replication
43        // group (that is, different from any other server id being used by any master or slave)
44        props
45            .entry("server.id".to_owned())
46            .or_insert(rand::rng().random_range(1..u32::MAX).to_string());
47        Ok(())
48    }
49
50    fn check_alter_with_options(
51        props: &BTreeMap<String, String>,
52        secret_refs: &BTreeMap<String, PbSecretRef>,
53    ) -> ConnectorResult<()> {
54        check_common_with_options(props)?;
55        check_alter_with_options_inner(
56            Self::CDC_CONNECTOR_NAME,
57            props,
58            secret_refs,
59            &[
60                "cdc.source.wait.streaming.start.timeout",
61                "debezium.max.queue.size",
62                "debezium.queue.memory.ratio",
63                "hostname",
64                "port",
65                "password",
66            ],
67        )
68    }
69}
risingwave_connector/source/cdc/mysql.rs

risingwave_connector/source/cdc/
mysql.rs
