Repository: risingwavelabs/risingwave
File: src/connector/src/source/cdc/enumerator/mod.rs (Rust)

Code before:
    async fn query_binlog_files(&self) -> ConnectorResult<Vec<(String, u64)>> {
        // Extract connection parameters from CDC properties
        let hostname = self
            .properties
            .get("hostname")
            .ok_or_else(|| anyhow::anyhow!("hostname not found in CDC properties"))?;
        let port = self
            .properties
            .get("port")
            .ok_or_else(|| anyhow::anyhow!("port not found in CDC properties"))?;
        let user = self
            .properties
            .get("username")
            .ok_or_else(|| anyhow::anyhow!("username not found in CDC properties"))?;
        let password = self
            .properties
            .get("password")
            .ok_or_else(|| anyhow::anyhow!("password not found in CDC properties"))?;
        let database = self
            .properties
            .get("database.name")
            .ok_or_else(|| anyhow::anyhow!("database.name not found in CDC properties"))?;

        // Build MySQL connection string
        let mysql_url = format!(
            "mysql://{}:{}@{}:{}/{}",
            user, password, hostname, port, database
        );

        // Create MySQL connection
        let pool = mysql_async::Pool::new(mysql_url.as_str());
        let mut conn = pool
            .get_conn()
            .await
            .context("Failed to connect to MySQL")?;

        // Query binlog files using SHOW BINARY LOGS
        // Note: MySQL 8.0+ returns 3 columns: Log_name, File_size, Encrypted
        let query_result: Vec<(String, u64)> = conn
            .query_map(
                "SHOW BINARY LOGS",
                |(log_name, file_size, _encrypted): (String, u64, String)| (log_name, file_size),
            )
            .await
            .context("Failed to execute SHOW BINARY LOGS")?;

        drop(conn);
        pool.disconnect().await.ok();

        Ok(query_result)
    }

Review comment (refactor):
How about the SSL related configs? https://github.com/risingwavelabs/risingwave/blob/d052368d0f54631b5649dce648b70ac323634954/src/connector/src/source/cdc/external/mysql.rs#L448

Can we extract the common codes to set up mysql connection based on the with properties? If I search for `mysql_async::Pool::new` in our code bases, there are many occurrences and I think it is easy to miss some configs if we don't unify them.

Code after:
        }
        Ok(())
    }

    /// Extract sequence number from binlog filename
    /// e.g., "binlog.000001" -> Some(1), "mysql-bin.000123" -> Some(123)
    fn extract_binlog_seq(filename: &str) -> Option<u64> {
        filename.rsplit('.').next()?.parse::<u64>().ok()
    }

    /// Query binlog files from MySQL, returns Vec<(filename, size)>
    async fn query_binlog_files(&self) -> ConnectorResult<Vec<(String, u64)>> {
        // Extract connection parameters from CDC properties
        let hostname = self
            .properties
            .get("hostname")
            .ok_or_else(|| anyhow::anyhow!("hostname not found in CDC properties"))?;
        let port = self
            .properties
            .get("port")
            .ok_or_else(|| anyhow::anyhow!("port not found in CDC properties"))?
            .parse::<u16>()
            .context("failed to parse port as u16")?;
        let username = self
            .properties
            .get("username")
            .ok_or_else(|| anyhow::anyhow!("username not found in CDC properties"))?;
        let password = self
            .properties
            .get("password")
            .ok_or_else(|| anyhow::anyhow!("password not found in CDC properties"))?;
        let database = self
            .properties
            .get("database.name")
            .ok_or_else(|| anyhow::anyhow!("database.name not found in CDC properties"))?;

        // Get SSL mode configuration (default to Disabled if not specified)
        let ssl_mode = self
            .properties
            .get("ssl.mode")
            .and_then(|s| serde_json::from_value(serde_json::Value::String(s.clone())).ok())
            .unwrap_or(SslMode::Preferred);

        // Build MySQL connection pool with proper SSL configuration
        let pool =
            build_mysql_connection_pool(hostname, port, username, password, database, ssl_mode);
        let mut conn = pool
            .get_conn()
            .await
            .context("Failed to connect to MySQL")?;