[Hotfix][CDC] Fix chunk start/end parameter type error (#4777)

Incorrect wrapping as Array<Array> types, but only Array type required

Author: hailin0

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/CompletedSnapshotSplitInfo.java

@@ -30,16 +30,16 @@ public class CompletedSnapshotSplitInfo implements Serializable {
     private final String splitId;
     private final TableId tableId;
     private final SeaTunnelRowType splitKeyType;
-    private final Object splitStart;
-    private final Object splitEnd;
+    private final Object[] splitStart;
+    private final Object[] splitEnd;
     private final Offset watermark;
 
     public CompletedSnapshotSplitInfo(
             String splitId,
             TableId tableId,
             SeaTunnelRowType splitKeyType,
-            Object splitStart,
-            Object splitEnd,
+            Object[] splitStart,
+            Object[] splitEnd,
             Offset watermark) {
         this.splitId = splitId;
         this.tableId = tableId;

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/split/SnapshotSplit.java

@@ -28,17 +28,17 @@ public class SnapshotSplit extends SourceSplitBase {
     private static final long serialVersionUID = 1L;
     private final TableId tableId;
     private final SeaTunnelRowType splitKeyType;
-    private final Object splitStart;
-    private final Object splitEnd;
+    private final Object[] splitStart;
+    private final Object[] splitEnd;
 
     private final Offset highWatermark;
 
     public SnapshotSplit(
             String splitId,
             TableId tableId,
             SeaTunnelRowType splitKeyType,
-            Object splitStart,
-            Object splitEnd,
+            Object[] splitStart,
+            Object[] splitEnd,
             Offset highWatermark) {
         super(splitId);
         this.tableId = tableId;

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/eumerator/MySqlChunkSplitter.java

@@ -295,8 +295,10 @@ private SnapshotSplit createSnapshotSplit(
             Object chunkStart,
             Object chunkEnd) {
         // currently, we only support single split column
+        Object[] splitStart = chunkStart == null ? null : new Object[] {chunkStart};
+        Object[] splitEnd = chunkEnd == null ? null : new Object[] {chunkEnd};
         return new SnapshotSplit(
-                splitId(tableId, chunkId), tableId, splitKeyType, chunkStart, chunkEnd, null);
+                splitId(tableId, chunkId), tableId, splitKeyType, splitStart, splitEnd, null);
     }
 
     // ------------------------------------------------------------------------------------------

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/scan/MySqlSnapshotSplitReadTask.java

@@ -205,8 +205,8 @@ private void createDataEventsForTable(
                                 selectSql,
                                 snapshotSplit.getSplitStart() == null,
                                 snapshotSplit.getSplitEnd() == null,
-                                new Object[] {snapshotSplit.getSplitStart()},
-                                new Object[] {snapshotSplit.getSplitEnd()},
+                                snapshotSplit.getSplitStart(),
+                                snapshotSplit.getSplitEnd(),
                                 snapshotSplit.getSplitKeyType().getTotalFields(),
                                 connectorConfig.getSnapshotFetchSize());
                 ResultSet rs = selectStatement.executeQuery()) {

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/eumerator/SqlServerChunkSplitter.java

@@ -292,8 +292,10 @@ private SnapshotSplit createSnapshotSplit(
             Object chunkStart,
             Object chunkEnd) {
         // currently, we only support single split column
+        Object[] splitStart = chunkStart == null ? null : new Object[] {chunkStart};
+        Object[] splitEnd = chunkEnd == null ? null : new Object[] {chunkEnd};
         return new SnapshotSplit(
-                splitId(tableId, chunkId), tableId, splitKeyType, chunkStart, chunkEnd, null);
+                splitId(tableId, chunkId), tableId, splitKeyType, splitStart, splitEnd, null);
     }
 
     // ------------------------------------------------------------------------------------------

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/source/reader/fetch/scan/SqlServerSnapshotSplitReadTask.java

@@ -195,8 +195,8 @@ private void createDataEventsForTable(
                                 selectSql,
                                 snapshotSplit.getSplitStart() == null,
                                 snapshotSplit.getSplitEnd() == null,
-                                new Object[] {snapshotSplit.getSplitStart()},
-                                new Object[] {snapshotSplit.getSplitEnd()},
+                                snapshotSplit.getSplitStart(),
+                                snapshotSplit.getSplitEnd(),
                                 snapshotSplit.getSplitKeyType().getTotalFields(),
                                 connectorConfig.getSnapshotFetchSize());
                 ResultSet rs = selectStatement.executeQuery()) {