fix kinesis property

risingwavelabs · Nov 27, 2023 · bcbe788 · bcbe788
1 parent 97eb91c
commit bcbe788
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 2 deletions.
diff --git a/src/connector/src/lib.rs b/src/connector/src/lib.rs
@@ -31,6 +31,7 @@
 #![feature(if_let_guard)]
 #![feature(iterator_try_collect)]
 #![feature(try_blocks)]
+#![feature(let_else)]
 
 use std::time::Duration;
 

diff --git a/src/connector/src/source/kinesis/mod.rs b/src/connector/src/source/kinesis/mod.rs
@@ -33,7 +33,7 @@ pub struct KinesisProperties {
     pub scan_startup_mode: Option<String>,
 
     #[serde(rename = "scan.startup.timestamp.millis")]
-    pub timestamp_offset: Option<i64>,
+    pub timestamp_offset: Option<String>,
 
     #[serde(flatten)]
     pub common: KinesisCommon,

diff --git a/src/connector/src/source/kinesis/source/reader.rs b/src/connector/src/source/kinesis/source/reader.rs
@@ -72,7 +72,12 @@ impl SplitReader for KinesisSplitReader {
                     "latest" => KinesisOffset::Latest,
                     "timestamp" => {
                         if let Some(ts) = &properties.timestamp_offset {
-                            KinesisOffset::Timestamp(*ts)
+                            let Ok(ts_i) = ts.parse() else {
+                                return Err(anyhow!(
+                                    format! {"cannot parse timestamp offset {ts} into i64"}
+                                ));
+                            };
+                            KinesisOffset::Timestamp(ts_i)
                         } else {
                             return Err(anyhow!("scan.startup.timestamp.millis is required"));
                         }