add test

Author: tabVersion

src/connector/src/source/kinesis/enumerator/client.rs

@@ -76,6 +76,7 @@ impl SplitEnumerator for KinesisSplitEnumerator {
             .into_iter()
             .map(|x| KinesisSplit {
                 shard_id: x.shard_id().unwrap_or_default().to_string().into(),
+                // handle start with position in reader part
                 start_position: KinesisOffset::None,
                 end_position: KinesisOffset::None,
             })

src/connector/src/source/kinesis/source/reader.rs

@@ -88,7 +88,10 @@ impl SplitReader for KinesisSplitReader {
             },
             start_position => start_position.to_owned(),
         };
-        if start_position != KinesisOffset::SequenceNumber && properties.seq_offset.is_some() {
+
+        if !matches!(start_position, KinesisOffset::SequenceNumber(_))
+            && properties.seq_offset.is_some()
+        {
             return Err(
                 anyhow!("scan.startup.mode need to be set to 'sequence_number' if you want to start with a specific sequence number")
             );
@@ -280,6 +283,41 @@ mod tests {
     use crate::common::KinesisCommon;
     use crate::source::kinesis::split::KinesisSplit;
 
+    #[tokio::test]
+    async fn test_reject_redundant_seq_props() {
+        let properties = KinesisProperties {
+            common: KinesisCommon {
+                assume_role_arn: None,
+                credentials_access_key: None,
+                credentials_secret_access_key: None,
+                stream_name: "kinesis_debug".to_string(),
+                stream_region: "cn-northwest-1".to_string(),
+                endpoint: None,
+                session_token: None,
+                assume_role_external_id: None,
+            },
+
+            scan_startup_mode: None,
+            seq_offset: Some(
+                // redundant seq number
+                "49629139817504901062972448413535783695568426186596941842".to_string(),
+            ),
+        };
+        let client = KinesisSplitReader::new(
+            properties,
+            vec![SplitImpl::Kinesis(KinesisSplit {
+                shard_id: "shardId-000000000001".to_string().into(),
+                start_position: KinesisOffset::Earliest,
+                end_position: KinesisOffset::None,
+            })],
+            Default::default(),
+            Default::default(),
+            None,
+        )
+        .await;
+        assert!(client.is_err());
+    }
+
     #[tokio::test]
     #[ignore]
     async fn test_single_thread_kinesis_reader() -> Result<()> {