[Improve][CDC] Optimize jdbc fetch-size options (#4352)

Author: hailin0

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

@@ -206,7 +206,7 @@ private void createDataEventsForTable(
                                 new Object[] {snapshotSplit.getSplitStart()},
                                 new Object[] {snapshotSplit.getSplitEnd()},
                                 snapshotSplit.getSplitKeyType().getTotalFields(),
-                                connectorConfig.getQueryFetchSize());
+                                connectorConfig.getSnapshotFetchSize());
                 ResultSet rs = selectStatement.executeQuery()) {
 
             ColumnUtils.ColumnArray columnArray = ColumnUtils.toArray(rs, table);

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/utils/MySqlUtils.java

@@ -344,8 +344,10 @@ private static PreparedStatement initStatement(JdbcConnection jdbc, String sql,
         final PreparedStatement statement =
                 connection.prepareStatement(
                         sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
-        if (fetchSize == 0) {
+        if (fetchSize <= 0) {
             statement.setFetchSize(Integer.MIN_VALUE);
+        } else {
+            statement.setFetchSize(fetchSize);
         }
         return statement;
     }

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

@@ -197,7 +197,7 @@ private void createDataEventsForTable(
                                 new Object[] {snapshotSplit.getSplitStart()},
                                 new Object[] {snapshotSplit.getSplitEnd()},
                                 snapshotSplit.getSplitKeyType().getTotalFields(),
-                                connectorConfig.getQueryFetchSize());
+                                connectorConfig.getSnapshotFetchSize());
                 ResultSet rs = selectStatement.executeQuery()) {
 
             ColumnUtils.ColumnArray columnArray = ColumnUtils.toArray(rs, table);