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

Code before:
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt::Debug;
use std::sync::Arc;

use futures::future::select_all;
use itertools::Itertools;
use reqwest::{Method, Url};
use serde::de::DeserializeOwned;
use thiserror_ext::AsReport as _;

use super::util::*;
use crate::schema::{invalid_option_error, InvalidOptionError};

pub const SCHEMA_REGISTRY_USERNAME: &str = "schema.registry.username";
pub const SCHEMA_REGISTRY_PASSWORD: &str = "schema.registry.password";
pub const SCHEMA_REGISTRY_CA: &str = "schema.registry.ca";

#[derive(Debug, Clone, Default)]
pub struct SchemaRegistryAuth {
    username: Option<String>,
    password: Option<String>,
    ca_option: Option<String>,
}

impl From<&HashMap<String, String>> for SchemaRegistryAuth {
    fn from(props: &HashMap<String, String>) -> Self {
        SchemaRegistryAuth {
            username: props.get(SCHEMA_REGISTRY_USERNAME).cloned(),
            password: props.get(SCHEMA_REGISTRY_PASSWORD).cloned(),
            ca_option: props.get(SCHEMA_REGISTRY_CA).cloned(),
        }
    }
}

impl From<&BTreeMap<String, String>> for SchemaRegistryAuth {
    fn from(props: &BTreeMap<String, String>) -> Self {
        SchemaRegistryAuth {
            username: props.get(SCHEMA_REGISTRY_USERNAME).cloned(),
            password: props.get(SCHEMA_REGISTRY_PASSWORD).cloned(),
            ca_option: props.get(SCHEMA_REGISTRY_CA).cloned(),
        }

Review comment (refactor):
In `src/frontend/src/handler/create_source.rs` there are 3 places that need:
```
try_consume_string_from_options(format_encode_options, SCHEMA_REGISTRY_CA);
```
to avoid warning about unrecognized option (which is actually recognized).

(I am trying to refactor those to make future updates easier.)

Code after:
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::collections::HashSet;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;

use futures::future::select_all;
use itertools::Itertools;
use reqwest::{Method, Url};
use serde::Deserialize;
use serde::de::DeserializeOwned;
use thiserror_ext::AsReport as _;
use tokio_retry::Retry;
use tokio_retry::strategy::{ExponentialBackoff, jitter};

use super::util::*;
use crate::connector_common::ConfluentSchemaRegistryConnection;
use crate::schema::{InvalidOptionError, invalid_option_error};
use crate::with_options::Get;

pub const SCHEMA_REGISTRY_USERNAME: &str = "schema.registry.username";
pub const SCHEMA_REGISTRY_PASSWORD: &str = "schema.registry.password";
pub const SCHEMA_REGISTRY_CA_PEM_PATH: &str = "schema.registry.ca_pem_path";

pub const SCHEMA_REGISTRY_MAX_DELAY_KEY: &str = "schema.registry.max.delay.sec";
pub const SCHEMA_REGISTRY_BACKOFF_DURATION_KEY: &str = "schema.registry.backoff.duration.ms";
pub const SCHEMA_REGISTRY_BACKOFF_FACTOR_KEY: &str = "schema.registry.backoff.factor";
pub const SCHEMA_REGISTRY_RETRIES_MAX_KEY: &str = "schema.registry.retries.max";

const DEFAULT_MAX_DELAY_SEC: u32 = 3;
const DEFAULT_BACKOFF_DURATION_MS: u64 = 100;
const DEFAULT_BACKOFF_FACTOR: u64 = 2;
const DEFAULT_RETRIES_MAX: usize = 3;

#[derive(Debug, Clone)]
struct SchemaRegistryRetryConfig {
    pub max_delay_sec: u32,
    pub backoff_duration_ms: u64,
    pub backoff_factor: u64,
    pub retries_max: usize,
}