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

Code before:

impl SourceProperties for PulsarProperties {
    type Split = PulsarSplit;
    type SplitEnumerator = PulsarSplitEnumerator;
    type SplitReader = PulsarSplitReader;

    const SOURCE_NAME: &'static str = PULSAR_CONNECTOR;
}

impl crate::source::UnknownFields for PulsarProperties {
    fn unknown_fields(&self) -> HashMap<String, String> {
        self.unknown_fields.clone()
    }
}

impl EnforceSecret for PulsarProperties {
    fn enforce_secret<'a>(prop_iter: impl Iterator<Item = &'a str>) -> Result<(), ConnectorError> {
        for prop in prop_iter {
            PulsarCommon::enforce_one(prop)?;
        }
        Ok(())
    }
}

impl EnforceSecret for PulsarConsumerOptions {}

#[derive(Clone, Debug, Deserialize, WithOptions)]
#[serde_as]
pub struct PulsarConsumerOptions {
    #[serde(rename = "pulsar.read_compacted")]
    #[serde_as(as = "Option<DisplayFromStr>")]
    pub read_compacted: Option<bool>,
}

#[derive(Clone, Debug, Deserialize, WithOptions)]
#[serde_as]
pub struct PulsarProperties {
    #[serde(rename = "scan.startup.mode", alias = "pulsar.scan.startup.mode")]
    pub scan_startup_mode: Option<String>,

    #[serde(
        rename = "scan.startup.timestamp.millis",
        alias = "pulsar.time.offset",
        alias = "scan.startup.timestamp_millis"
    )]
    pub time_offset: Option<String>,

    #[serde(flatten)]
    pub common: PulsarCommon,

    #[serde(flatten)]

Review comment (security):
The empty `EnforceSecret` implementation for `PulsarConsumerOptions` lacks documentation explaining why no secrets need enforcement. Add a comment clarifying that consumer options don't contain sensitive data.
```suggestion

// PulsarConsumerOptions does not contain any sensitive data, so no secret enforcement is necessary.
```

Code after:

pub const PULSAR_CONNECTOR: &str = "pulsar";

impl SourceProperties for PulsarProperties {
    type Split = PulsarSplit;
    type SplitEnumerator = PulsarSplitEnumerator;
    type SplitReader = PulsarSplitReader;

    const SOURCE_NAME: &'static str = PULSAR_CONNECTOR;
}

impl crate::source::UnknownFields for PulsarProperties {
    fn unknown_fields(&self) -> HashMap<String, String> {
        self.unknown_fields.clone()
    }
}

impl EnforceSecret for PulsarProperties {
    fn enforce_secret<'a>(prop_iter: impl Iterator<Item = &'a str>) -> Result<(), ConnectorError> {
        for prop in prop_iter {
            PulsarCommon::enforce_one(prop)?;
        }
        Ok(())
    }
}

impl EnforceSecret for PulsarConsumerOptions {}

#[derive(Clone, Debug, Deserialize, WithOptions)]
#[serde_as]
pub struct PulsarConsumerOptions {
    #[serde(
        rename = "pulsar.read_compacted",
        default,
        deserialize_with = "deserialize_optional_bool_from_string"
    )]
    pub read_compacted: Option<bool>,
}

#[derive(Clone, Debug, Deserialize, WithOptions)]
#[serde_as]
pub struct PulsarProperties {
    #[serde(rename = "scan.startup.mode", alias = "pulsar.scan.startup.mode")]
    pub scan_startup_mode: Option<String>,

    #[serde(
        rename = "scan.startup.timestamp.millis",
        alias = "pulsar.time.offset",
        alias = "scan.startup.timestamp_millis"
    )]
    pub time_offset: Option<String>,