[FLINK-39391][cdc-connector] Propagate scan.snapshot.fetch.size to Debezium properties in Oracle, SqlServer, DB2, and Postgres connectors

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/main/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactory.java

@@ -75,6 +75,8 @@ public Db2SourceConfig create(int subtask) {
                 throw new UnsupportedOperationException();
         }
 
+        props.setProperty("query.fetch.size", String.valueOf(fetchSize));
+
         if (dbzProperties != null) {
             props.putAll(dbzProperties);
         }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-db2-cdc/src/test/java/org/apache/flink/cdc/connectors/db2/source/config/Db2SourceConfigFactoryTest.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.db2.source.config;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link Db2SourceConfigFactory}. */
+class Db2SourceConfigFactoryTest {
+
+    @Test
+    void testFetchSizePropagatedToDebeziumProperties() {
+        Db2SourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+
+        Db2SourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("5000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(5000);
+    }
+
+    @Test
+    void testDefaultFetchSizePropagatedToDebeziumProperties() {
+        Db2SourceConfigFactory factory = createFactory();
+
+        Db2SourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("1024");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(1024);
+    }
+
+    @Test
+    void testDebeziumPropertiesCanOverrideFetchSize() {
+        Db2SourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+        Properties dbzProps = new Properties();
+        dbzProps.setProperty("query.fetch.size", "8000");
+        factory.debeziumProperties(dbzProps);
+
+        Db2SourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("8000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(8000);
+    }
+
+    private static Db2SourceConfigFactory createFactory() {
+        Db2SourceConfigFactory factory = new Db2SourceConfigFactory();
+        factory.hostname("localhost");
+        factory.port(50000);
+        factory.databaseList("myDB");
+        factory.username("user");
+        factory.password("password");
+        factory.startupOptions(StartupOptions.initial());
+        return factory;
+    }
+}

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/main/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactory.java

@@ -103,6 +103,8 @@ public OracleSourceConfig create(int subtaskId) {
             props.setProperty("table.include.list", String.join(",", tableList));
         }
 
+        props.setProperty("query.fetch.size", String.valueOf(fetchSize));
+
         // override the user-defined debezium properties
         if (dbzProperties != null) {
             props.putAll(dbzProperties);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/src/test/java/org/apache/flink/cdc/connectors/oracle/source/config/OracleSourceConfigFactoryTest.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.oracle.source.config;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link OracleSourceConfigFactory}. */
+class OracleSourceConfigFactoryTest {
+
+    @Test
+    void testFetchSizePropagatedToDebeziumProperties() {
+        OracleSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+
+        OracleSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("5000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(5000);
+    }
+
+    @Test
+    void testDefaultFetchSizePropagatedToDebeziumProperties() {
+        OracleSourceConfigFactory factory = createFactory();
+
+        OracleSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("1024");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(1024);
+    }
+
+    @Test
+    void testDebeziumPropertiesCanOverrideFetchSize() {
+        OracleSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+        Properties dbzProps = new Properties();
+        dbzProps.setProperty("query.fetch.size", "8000");
+        factory.debeziumProperties(dbzProps);
+
+        OracleSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("8000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(8000);
+    }
+
+    private static OracleSourceConfigFactory createFactory() {
+        OracleSourceConfigFactory factory = new OracleSourceConfigFactory();
+        factory.hostname("localhost");
+        factory.port(1521);
+        factory.databaseList("MYDB");
+        factory.username("user");
+        factory.password("password");
+        factory.startupOptions(StartupOptions.initial());
+        return factory;
+    }
+}

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java

@@ -100,6 +100,8 @@ public PostgresSourceConfig create(int subtaskId) {
             props.setProperty("table.include.list", String.join(",", tableList));
         }
 
+        props.setProperty("snapshot.fetch.size", String.valueOf(fetchSize));
+
         // override the user-defined debezium properties
         if (dbzProperties != null) {
             props.putAll(dbzProperties);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactoryTest.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.config;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PostgresSourceConfigFactory}. */
+class PostgresSourceConfigFactoryTest {
+
+    @Test
+    void testFetchSizePropagatedToDebeziumProperties() {
+        PostgresSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+
+        PostgresSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("snapshot.fetch.size")).isEqualTo("5000");
+        assertThat(config.getDbzConnectorConfig().getSnapshotFetchSize()).isEqualTo(5000);
+    }
+
+    @Test
+    void testDefaultFetchSizePropagatedToDebeziumProperties() {
+        PostgresSourceConfigFactory factory = createFactory();
+
+        PostgresSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("snapshot.fetch.size")).isEqualTo("1024");
+        assertThat(config.getDbzConnectorConfig().getSnapshotFetchSize()).isEqualTo(1024);
+    }
+
+    @Test
+    void testDebeziumPropertiesCanOverrideFetchSize() {
+        PostgresSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+        Properties dbzProps = new Properties();
+        dbzProps.setProperty("snapshot.fetch.size", "8000");
+        factory.debeziumProperties(dbzProps);
+
+        PostgresSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("snapshot.fetch.size")).isEqualTo("8000");
+        assertThat(config.getDbzConnectorConfig().getSnapshotFetchSize()).isEqualTo(8000);
+    }
+
+    private static PostgresSourceConfigFactory createFactory() {
+        PostgresSourceConfigFactory factory = new PostgresSourceConfigFactory();
+        factory.hostname("localhost");
+        factory.port(5432);
+        factory.database("myDB");
+        factory.username("user");
+        factory.password("password");
+        factory.startupOptions(StartupOptions.initial());
+        return factory;
+    }
+}

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactory.java

@@ -75,6 +75,8 @@ public SqlServerSourceConfig create(int subtask) {
                 throw new UnsupportedOperationException();
         }
 
+        props.setProperty("query.fetch.size", String.valueOf(fetchSize));
+
         if (dbzProperties != null) {
             props.putAll(dbzProperties);
         }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/config/SqlServerSourceConfigFactoryTest.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.sqlserver.source.config;
+
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link SqlServerSourceConfigFactory}. */
+class SqlServerSourceConfigFactoryTest {
+
+    @Test
+    void testFetchSizePropagatedToDebeziumProperties() {
+        SqlServerSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+
+        SqlServerSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("5000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(5000);
+    }
+
+    @Test
+    void testDefaultFetchSizePropagatedToDebeziumProperties() {
+        SqlServerSourceConfigFactory factory = createFactory();
+
+        SqlServerSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("1024");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(1024);
+    }
+
+    @Test
+    void testDebeziumPropertiesCanOverrideFetchSize() {
+        SqlServerSourceConfigFactory factory = createFactory();
+        factory.fetchSize(5000);
+        Properties dbzProps = new Properties();
+        dbzProps.setProperty("query.fetch.size", "8000");
+        factory.debeziumProperties(dbzProps);
+
+        SqlServerSourceConfig config = factory.create(0);
+
+        assertThat(config.getDbzProperties().getProperty("query.fetch.size")).isEqualTo("8000");
+        assertThat(config.getDbzConnectorConfig().getQueryFetchSize()).isEqualTo(8000);
+    }
+
+    private static SqlServerSourceConfigFactory createFactory() {
+        SqlServerSourceConfigFactory factory = new SqlServerSourceConfigFactory();
+        factory.hostname("localhost");
+        factory.port(1433);
+        factory.databaseList("myDB");
+        factory.username("user");
+        factory.password("password");
+        factory.startupOptions(StartupOptions.initial());
+        return factory;
+    }
+}