Repository: risingwavelabs/risingwave
File: src/connector/src/schema/schema_registry/client.rs (Rust)

Code before:
    context: String,
}

type SrResult<T> = Result<T, ConcurrentRequestError>;

impl Client {
    pub(crate) fn new(
        url: Vec<Url>,
        client_config: &SchemaRegistryAuth,
    ) -> Result<Self, InvalidOptionError> {
        let valid_urls = url
            .iter()
            .map(|url| (url.cannot_be_a_base(), url))
            .filter(|(x, _)| !*x)
            .map(|(_, url)| url.clone())
            .collect_vec();
        if valid_urls.is_empty() {
            return Err(invalid_option_error!("non-base: {}", url.iter().join(" ")));
        } else {
            tracing::debug!(
                "schema registry client will use url {:?} to connect",
                valid_urls
            );
        }

        // `unwrap` as the builder is not affected by any input right now
        let mut client_builder = reqwest::Client::builder();
        if let Some(ca_path) = client_config.ca_option.as_ref() {
            if ca_path.eq_ignore_ascii_case("ignore") {
                client_builder = client_builder.danger_accept_invalid_certs(true);
            } else {
                client_builder = client_builder.add_root_certificate(
                    reqwest::Certificate::from_pem(&std::fs::read(ca_path).map_err(|e| {
                        invalid_option_error!("read ca file error: {}", e.as_report())
                    })?)
                    .map_err(|e| invalid_option_error!("parse ca file error: {}", e.as_report()))?,
                );
            }
        }

        let inner = client_builder.build().map_err(|e| {
            invalid_option_error!("build schema registry client error: {}", e.as_report())
        })?;

        Ok(Client {
            inner,
            url: valid_urls,
            username: client_config.username.clone(),
            password: client_config.password.clone(),
        })
    }

Review comment (suggestion):
This comment is no longer accurate and shall be removed. It is now affected by input (ca) and we are no longer using `unwrap`.

Code after:

    retry_config: SchemaRegistryRetryConfig,
}

impl<T: Get> From<&T> for SchemaRegistryConfig {
    fn from(props: &T) -> Self {
        SchemaRegistryConfig {
            username: props.get(SCHEMA_REGISTRY_USERNAME).cloned(),
            password: props.get(SCHEMA_REGISTRY_PASSWORD).cloned(),
            ca_pem_path: props.get(SCHEMA_REGISTRY_CA_PEM_PATH).cloned(),

            retry_config: SchemaRegistryRetryConfig {
                max_delay_sec: props
                    .get(SCHEMA_REGISTRY_MAX_DELAY_KEY)
                    .and_then(|v| v.parse::<u32>().ok())
                    .unwrap_or(DEFAULT_MAX_DELAY_SEC),
                backoff_duration_ms: props
                    .get(SCHEMA_REGISTRY_BACKOFF_DURATION_KEY)
                    .and_then(|v| v.parse::<u64>().ok())
                    .unwrap_or(DEFAULT_BACKOFF_DURATION_MS),
                backoff_factor: props
                    .get(SCHEMA_REGISTRY_BACKOFF_FACTOR_KEY)
                    .and_then(|v| v.parse::<u64>().ok())
                    .unwrap_or(DEFAULT_BACKOFF_FACTOR),
                retries_max: props
                    .get(SCHEMA_REGISTRY_RETRIES_MAX_KEY)
                    .and_then(|v| v.parse::<usize>().ok())
                    .unwrap_or(DEFAULT_RETRIES_MAX),
            },
        }
    }
}

/// An client for communication with schema registry
#[derive(Debug)]
pub struct Client {
    inner: reqwest::Client,
    url: Vec<Url>,
    username: Option<String>,
    password: Option<String>,

    retry_config: SchemaRegistryRetryConfig,
}

#[derive(Debug, thiserror::Error)]
#[error("all request confluent registry all timeout, {context}\n{}", errs.iter().map(|e| format!("\t{}", e.as_report())).join("\n"))]
pub struct ConcurrentRequestError {
    errs: Vec<itertools::Either<RequestError, tokio::task::JoinError>>,
    context: String,
}