fix: reject kinesis source if the start up mode is not sequence number but sequence number is provided (#12048)

Author: tabVersion

e2e_test/source/basic/kinesis.slt

@@ -0,0 +1,11 @@
+# reject kinesis source if the start up mode is not sequence number but sequence number is provided
+statement error
+create source s ( a int, b varchar )
+with (
+  connector = 'kinesis',
+  stream = 'my_stream',
+  scan.startup.sequence_number = '4950',
+  aws.region = 'us-east-1',
+  kinesis.credentials.access = 'my_access_key_id',
+  kinesis.credentials.secret = 'my_secret_access_key',
+) format plain encode json;

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

@@ -89,6 +89,14 @@ impl SplitReader for KinesisSplitReader {
             start_position => start_position.to_owned(),
         };
 
+        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")
+            );
+        }
+
         let stream_name = properties.common.stream_name.clone();
         let client = properties.common.build_client().await?;
 
@@ -275,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<()> {