This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 308359e54 [FLINK-40007][connect/postgres] Fix snapshot fetch size conf
does not take effect (#4453)
308359e54 is described below
commit 308359e541271e54b0ebd9d66bd37663a474a689
Author: wudi <[email protected]>
AuthorDate: Tue Aug 4 14:04:32 2026 +0800
[FLINK-40007][connect/postgres] Fix snapshot fetch size conf does not take
effect (#4453)
* [FLINK-40007][postgres] Fix snapshot fetch size conf does not take effect
* [FLINK-40007][postgres] Fix testSnapshotFetchSize to use
sourceConfig.getFetchSize() instead of debezium getSnapshotFetchSize()
* code style
---
.../postgres/source/fetch/PostgresScanFetchTask.java | 8 +++++---
.../postgres/source/fetch/PostgresScanFetchTaskTest.java | 11 ++---------
2 files changed, 7 insertions(+), 12 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
index 1915aca80..d2d47d709 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTask.java
@@ -98,6 +98,7 @@ public class PostgresScanFetchTask extends
AbstractScanFetchTask {
PostgresSnapshotSplitReadTask snapshotSplitReadTask =
new PostgresSnapshotSplitReadTask(
+ (PostgresSourceConfig) ctx.getSourceConfig(),
ctx.getConnection(),
ctx.getDbzConnectorConfig(),
ctx.getDatabaseSchema(),
@@ -214,7 +215,7 @@ public class PostgresScanFetchTask extends
AbstractScanFetchTask {
LoggerFactory.getLogger(PostgresSnapshotSplitReadTask.class);
private final PostgresConnection jdbcConnection;
- private final PostgresConnectorConfig connectorConfig;
+ private final PostgresSourceConfig sourceConfig;
private final PostgresEventDispatcher<TableId> eventDispatcher;
private final SnapshotSplit snapshotSplit;
private final PostgresOffsetContext offsetContext;
@@ -223,6 +224,7 @@ public class PostgresScanFetchTask extends
AbstractScanFetchTask {
private final Clock clock;
public PostgresSnapshotSplitReadTask(
+ PostgresSourceConfig sourceConfig,
PostgresConnection jdbcConnection,
PostgresConnectorConfig connectorConfig,
PostgresSchema databaseSchema,
@@ -232,7 +234,7 @@ public class PostgresScanFetchTask extends
AbstractScanFetchTask {
SnapshotSplit snapshotSplit) {
super(connectorConfig, snapshotProgressListener);
this.jdbcConnection = jdbcConnection;
- this.connectorConfig = connectorConfig;
+ this.sourceConfig = sourceConfig;
this.snapshotProgressListener = snapshotProgressListener;
this.databaseSchema = databaseSchema;
this.eventDispatcher = eventDispatcher;
@@ -314,7 +316,7 @@ public class PostgresScanFetchTask extends
AbstractScanFetchTask {
snapshotSplit.getSplitStart(),
snapshotSplit.getSplitEnd(),
snapshotSplit.getSplitKeyType().getFieldCount(),
- connectorConfig.getSnapshotFetchSize());
+ sourceConfig.getFetchSize());
ResultSet rs = selectStatement.executeQuery()) {
ColumnUtils.ColumnArray columnArray = ColumnUtils.toArray(rs,
table);
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
index 8406b98bb..eb9834b17 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
@@ -55,7 +55,6 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
-import java.util.Properties;
import static org.assertj.core.api.Assertions.assertThat;
@@ -300,14 +299,10 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
PostgresSourceConfigFactory sourceConfigFactory =
getMockPostgresSourceConfigFactory(
customDatabase, schemaName, tableName, null, 10, true);
- Properties properties = new Properties();
- properties.setProperty("snapshot.fetch.size", "2");
- sourceConfigFactory.debeziumProperties(properties);
+ sourceConfigFactory.fetchSize(2);
PostgresSourceConfig sourceConfig = sourceConfigFactory.create(0);
PostgresDialect postgresDialect = new
PostgresDialect(sourceConfigFactory.create(0));
SnapshotSplit snapshotSplit = getSnapshotSplits(sourceConfig,
postgresDialect).get(0);
- PostgresSourceFetchTaskContext postgresSourceFetchTaskContext =
- new PostgresSourceFetchTaskContext(sourceConfig,
postgresDialect);
final String selectSql =
PostgresQueryUtils.buildSplitScanQuery(
snapshotSplit.getTableId(),
@@ -326,9 +321,7 @@ class PostgresScanFetchTaskTest extends PostgresTestBase {
snapshotSplit.getSplitStart(),
snapshotSplit.getSplitEnd(),
snapshotSplit.getSplitKeyType().getFieldCount(),
- postgresSourceFetchTaskContext
- .getDbzConnectorConfig()
- .getSnapshotFetchSize());
+ sourceConfig.getFetchSize());
ResultSet rs = selectStatement.executeQuery()) {
assertThat(rs.getFetchSize()).isEqualTo(2);
}
