Repository: risingwavelabs/risingwave
File: src/connector/src/connector_common/common.rs (Rust)

Code before:
    certificates: &str,
) -> ConnectorResult<Vec<rustls_pki_types::CertificateDer<'static>>> {
    let cert_bytes = if let Some(path) = certificates.strip_prefix("fs://") {
        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
    } else {
        certificates.as_bytes().to_owned()
    };

    CertificateDer::pem_slice_iter(&cert_bytes)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| anyhow!("Failed to parse certificates: {}", e).into())
}

pub(crate) fn load_private_key(
    certificate: &str,
) -> ConnectorResult<rustls_pki_types::PrivateKeyDer<'static>> {
    let cert_bytes = if let Some(path) = certificate.strip_prefix("fs://") {
        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
    } else {
        certificate.as_bytes().to_owned()
    };

    let cert = PrivatePkcs8KeyDer::pem_slice_iter(&cert_bytes)
        .next()
        .ok_or_else(|| anyhow!("No private key found"))?
        .map_err(|e| anyhow!("Failed to parse private key: {}", e))?;
    Ok(cert.into())
}

#[serde_as]
#[derive(Deserialize, Debug, Clone, WithOptions)]
pub struct MongodbCommon {
    /// The URL of `MongoDB`
    #[serde(rename = "mongodb.url")]
    pub connect_uri: String,
    /// The collection name where data should be written to or read from. For sinks, the format is
    /// `db_name.collection_name`. Data can also be written to dynamic collections, see `collection.name.field`
    /// for more information.
    #[serde(rename = "collection.name")]
    pub collection_name: String,
}

impl EnforceSecret for MongodbCommon {
    const ENFORCE_SECRET_PROPERTIES: Set<&'static str> = phf_set! {
        "mongodb.url"
    };
}

impl MongodbCommon {
    pub(crate) async fn build_client(&self) -> ConnectorResult<mongodb::Client> {
        let client = mongodb::Client::with_uri_str(&self.connect_uri).await?;

Review comment (bug):
```suggestion
        .context("Failed to parse private key")?;
```

Code after:
    certificates: &str,
) -> ConnectorResult<Vec<rustls_pki_types::CertificateDer<'static>>> {
    let cert_bytes = if let Some(path) = certificates.strip_prefix("fs://") {
        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
    } else {
        certificates.as_bytes().to_owned()
    };

    CertificateDer::pem_slice_iter(&cert_bytes)
        .collect::<Result<Vec<_>, _>>()
        .context("Failed to parse certificates")
        .map_err(Into::into)
}

pub(crate) fn load_private_key(
    certificate: &str,
) -> ConnectorResult<rustls_pki_types::PrivateKeyDer<'static>> {
    let cert_bytes = if let Some(path) = certificate.strip_prefix("fs://") {
        std::fs::read_to_string(path).map(|cert| cert.as_bytes().to_owned())?
    } else {
        certificate.as_bytes().to_owned()
    };

    let cert = PrivatePkcs8KeyDer::pem_slice_iter(&cert_bytes)
        .next()
        .ok_or_else(|| anyhow!("No private key found"))?
        .context("Failed to parse private key")?;
    Ok(cert.into())
}

#[serde_as]
#[derive(Deserialize, Debug, Clone, WithOptions)]
pub struct MongodbCommon {
    /// The URL of `MongoDB`
    #[serde(rename = "mongodb.url")]
    pub connect_uri: String,
    /// The collection name where data should be written to or read from. For sinks, the format is
    /// `db_name.collection_name`. Data can also be written to dynamic collections, see `collection.name.field`
    /// for more information.
    #[serde(rename = "collection.name")]
    pub collection_name: String,
}

impl EnforceSecret for MongodbCommon {
    const ENFORCE_SECRET_PROPERTIES: Set<&'static str> = phf_set! {
        "mongodb.url"
    };
}

impl MongodbCommon {
    pub(crate) async fn build_client(&self) -> ConnectorResult<mongodb::Client> {