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

Code before:
        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(),
        })
    }

    async fn concurrent_req<'a, T>(
        &'a self,
        method: Method,
        path: &'a [&'a (impl AsRef<str> + ?Sized + Debug + ToString)],
    ) -> SrResult<T>
    where
        T: DeserializeOwned + Send + Sync + 'static,
    {
        let mut fut_req = Vec::with_capacity(self.url.len());
        let mut errs = Vec::with_capacity(self.url.len());
        let ctx = Arc::new(SchemaRegistryCtx {
            username: self.username.clone(),
            password: self.password.clone(),
            client: self.inner.clone(),
            path: path.iter().map(|p| p.to_string()).collect_vec(),

Review comment (bug):
Avoid INTERPOLATION of source error messages (`{}` on `e.as_report()`)! Define a dedicated error to keep the source error:
```
#[derive(thiserror::Error)]
enum ClientBuildError {
  #[error(transparent)]
  InvalidOption(#[from] InvalidOptionError),
  #[error("read ca file error: {0}")]
  ReadFile(#[source] std::io::Error),
  #[error("parse ca file error: {0}")]
  ParsePrm(#[source] reqwest::Error),
  #[error("build schema registry client error: {0}")]
  Build(#[source] reqwest::Error),
}
```
So that:
* `Client::new` returns `ClientBuildError` rather than `InvalidOptionError`
* callers of `Client::new` may eventually transform `ClientBuildError` into a new variant of `SchemaFetchError`

cc @BugenZhao if there are simpler ways

Code after:
                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,
}

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

#[derive(thiserror::Error, Debug)]
pub enum SchemaRegistryClientError {
    #[error(transparent)]
    InvalidOption(#[from] InvalidOptionError),
    #[error("read ca file error: {0}")]
    ReadFile(#[source] std::io::Error),
    #[error("parse ca file error: {0}")]
    ParsePem(#[source] reqwest::Error),
    #[error("build schema registry client error: {0}")]
    Build(#[source] reqwest::Error),
}

impl TryFrom<&ConfluentSchemaRegistryConnection> for Client {
    type Error = InvalidOptionError;