Use serde-as

risingwavelabs · Oct 26, 2023 · ce18288 · ce18288
1 parent f440380
commit ce18288
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 2 deletions.
diff --git a/src/connector/src/source/pulsar/mod.rs b/src/connector/src/source/pulsar/mod.rs
@@ -19,6 +19,7 @@ pub mod topic;
 
 pub use enumerator::*;
 use serde::Deserialize;
+use serde_with::serde_as;
 pub use split::*;
 
 use self::source::reader::PulsarSplitReader;
@@ -36,6 +37,7 @@ impl SourceProperties for PulsarProperties {
 }
 
 #[derive(Clone, Debug, Deserialize)]
+#[serde_as]
 pub struct PulsarProperties {
     #[serde(rename = "scan.startup.mode", alias = "pulsar.scan.startup.mode")]
     pub scan_startup_mode: Option<String>,
@@ -46,7 +48,7 @@ pub struct PulsarProperties {
     #[serde(flatten)]
     pub common: PulsarCommon,
 
-    #[serde(rename = "iceberg.enabled", default)]
+    #[serde(rename = "iceberg.enabled")]
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub iceberg_loader_enabled: bool,
 

diff --git a/src/connector/src/source/pulsar/source/reader.rs b/src/connector/src/source/pulsar/source/reader.rs
@@ -65,7 +65,7 @@ impl SplitReader for PulsarSplitReader {
 
         tracing::debug!("creating consumer for pulsar split topic {}", topic,);
 
-        if props.is_iceberg_loader_enabled()
+        if props.iceberg_loader_enabled
             && matches!(split.start_offset, PulsarEnumeratorOffset::Earliest)
             && !topic.starts_with("non-persistent://")
         {