This is an automated email from the ASF dual-hosted git repository. leonard 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 5b28d1a57 [FLINK-35129][postgres] Introduce scan.lsn-commit.checkpoints-num-delay option to control LSN offset commit lazily 5b28d1a57 is described below commit 5b28d1a579919b29acac6acded46d9bee5596bde Author: Muhammet Orazov <916295+mora...@users.noreply.github.com> AuthorDate: Tue Jun 4 10:56:03 2024 +0200 [FLINK-35129][postgres] Introduce scan.lsn-commit.checkpoints-num-delay option to control LSN offset commit lazily This close #3349. --- .../docs/connectors/flink-sources/postgres-cdc.md | 40 +++++---- .../flink-connector-postgres-cdc/pom.xml | 8 +- .../postgres/source/PostgresDialect.java | 4 +- .../postgres/source/PostgresSourceBuilder.java | 6 ++ .../source/config/PostgresSourceConfig.java | 26 +++++- .../source/config/PostgresSourceConfigFactory.java | 12 ++- .../source/config/PostgresSourceOptions.java | 10 +++ .../source/reader/PostgresSourceReader.java | 21 ++++- .../postgres/table/PostgreSQLTableFactory.java | 6 +- .../postgres/table/PostgreSQLTableSource.java | 11 ++- .../cdc/connectors/postgres/PostgresTestBase.java | 1 + .../postgres/source/PostgresSourceITCase.java | 4 +- .../source/reader/PostgresSourceReaderTest.java | 94 ++++++++++++++++++++++ .../postgres/table/MockPostgreSQLTableSource.java | 3 +- .../postgres/table/PostgreSQLTableFactoryTest.java | 16 ++-- 15 files changed, 226 insertions(+), 36 deletions(-) diff --git a/docs/content/docs/connectors/flink-sources/postgres-cdc.md b/docs/content/docs/connectors/flink-sources/postgres-cdc.md index f839c4380..1e24f31d1 100644 --- a/docs/content/docs/connectors/flink-sources/postgres-cdc.md +++ b/docs/content/docs/connectors/flink-sources/postgres-cdc.md @@ -28,8 +28,7 @@ under the License. The Postgres CDC connector allows for reading snapshot data and incremental data from PostgreSQL database. This document describes how to setup the Postgres CDC connector to run SQL queries against PostgreSQL databases. -Dependencies ------------- +## Dependencies In order to setup the Postgres CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles. @@ -45,8 +44,7 @@ Download [flink-sql-connector-postgres-cdc-3.0.1.jar](https://repo1.maven.org/ma **Note:** Refer to [flink-sql-connector-postgres-cdc](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-postgres-cdc), more released versions will be available in the Maven central warehouse. -How to create a Postgres CDC table ----------------- +## How to create a Postgres CDC table The Postgres CDC table can be defined as following: @@ -76,8 +74,7 @@ CREATE TABLE shipments ( SELECT * FROM shipments; ``` -Connector Options ----------------- +## Connector Options <div class="highlight"> <table class="colwidths-auto docutils"> @@ -236,12 +233,29 @@ Connector Options so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true' </td> </tr> + <tr> + <td>scan.lsn-commit.checkpoints-num-delay</td> + <td>optional</td> + <td style="word-wrap: break-word;">3</td> + <td>Integer</td> + <td>The number of checkpoint delays before starting to commit the LSN offsets. <br> + The checkpoint LSN offsets will be committed in rolling fashion, the earliest checkpoint identifier will be committed first from the delayed checkpoints. + </td> + </tr> </tbody> </table> </div> <div> -Note: `slot.name` is recommended to set for different tables to avoid the potential `PSQLException: ERROR: replication slot "flink" is active for PID 974` error. See more [here](https://debezium.io/documentation/reference/2.0/connectors/postgresql.html#postgresql-property-slot-name). +### Notes + +#### `slot.name` option + +The `slot.name` is recommended to set for different tables to avoid the potential `PSQLException: ERROR: replication slot "flink" is active for PID 974` error. See more [here](https://debezium.io/documentation/reference/2.0/connectors/postgresql.html#postgresql-property-slot-name). + +#### `scan.lsn-commit.checkpoints-num-delay` option + +When consuming PostgreSQL logs, the LSN offset must be committed to trigger the log data cleanup for the corresponding slot. However, once the LSN offset is committed, earlier offsets become invalid. To ensure access to earlier LSN offsets for job recovery, we delay the LSN commit by `scan.lsn-commit.checkpoints-num-delay` (default value is `3`) checkpoints. This feature is available when config option `scan.incremental.snapshot.enabled` is set to true. ### Incremental Snapshot Options @@ -340,8 +354,7 @@ The following options is available only when `scan.incremental.snapshot.enabled= </table> </div> -Available Metadata ----------------- +## Available Metadata The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition. @@ -377,8 +390,7 @@ The following format metadata can be exposed as read-only (VIRTUAL) columns in a </tbody> </table> -Limitation --------- +## Limitation ### Can't perform checkpoint during scanning snapshot of tables when incremental snapshot is disabled @@ -417,8 +429,7 @@ CREATE TABLE products ( ); ``` -Features --------- +## Features ### Incremental Snapshot Reading (Experimental) @@ -522,8 +533,7 @@ public class PostgreSQLSourceExample { } ``` -Data Type Mapping ----------------- +## Data Type Mapping <div class="wy-table-responsive"> <table class="colwidths-auto docutils"> diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml index b0763b1d7..4fc75d0ca 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml @@ -164,10 +164,16 @@ limitations under the License. <version>${json-path.version}</version> <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-test-utils</artifactId> + <version>${flink.version}</version> + <scope>test</scope> + </dependency> <!-- tests will have log4j as the default logging framework available --> </dependencies> -</project> \ No newline at end of file +</project> diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java index 5e83f6685..25a5186d1 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresDialect.java @@ -63,13 +63,11 @@ import static io.debezium.connector.postgresql.Utils.currentOffset; /** The dialect for Postgres. */ public class PostgresDialect implements JdbcDataSourceDialect { private static final long serialVersionUID = 1L; - private static final String CONNECTION_NAME = "postgres-cdc-connector"; + private final PostgresSourceConfig sourceConfig; private transient Tables.TableFilter filters; - private transient CustomPostgresSchema schema; - @Nullable private PostgresStreamFetchTask streamFetchTask; public PostgresDialect(PostgresSourceConfig sourceConfig) { diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java index 6fa882f56..ee991a70f 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceBuilder.java @@ -280,6 +280,12 @@ public class PostgresSourceBuilder<T> { return this; } + /** Set the {@code LSN} checkpoints delay number for Postgres to commit the offsets. */ + public PostgresSourceBuilder<T> lsnCommitCheckpointsDelay(int lsnCommitDelay) { + this.configFactory.setLsnCommitCheckpointsDelay(lsnCommitDelay); + return this; + } + /** * Build the {@link PostgresIncrementalSource}. * diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java index c20e57fab..a4e30da11 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java @@ -37,6 +37,7 @@ public class PostgresSourceConfig extends JdbcSourceConfig { private static final long serialVersionUID = 1L; private final int subtaskId; + private final int lsnCommitCheckpointsDelay; public PostgresSourceConfig( int subtaskId, @@ -64,7 +65,8 @@ public class PostgresSourceConfig extends JdbcSourceConfig { int connectionPoolSize, @Nullable String chunkKeyColumn, boolean skipSnapshotBackfill, - boolean isScanNewlyAddedTableEnabled) { + boolean isScanNewlyAddedTableEnabled, + int lsnCommitCheckpointsDelay) { super( startupOptions, databaseList, @@ -92,14 +94,34 @@ public class PostgresSourceConfig extends JdbcSourceConfig { skipSnapshotBackfill, isScanNewlyAddedTableEnabled); this.subtaskId = subtaskId; + this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay; } + /** + * Returns {@code subtaskId} value. + * + * @return subtask id + */ public int getSubtaskId() { return subtaskId; } + /** + * Returns {@code lsnCommitCheckpointsDelay} value. + * + * @return lsn commit checkpoint delay + */ + public int getLsnCommitCheckpointsDelay() { + return this.lsnCommitCheckpointsDelay; + } + + /** + * Returns the slot name for backfill task. + * + * @return backfill task slot name + */ public String getSlotNameForBackfillTask() { - return getDbzProperties().getProperty(SLOT_NAME.name()) + "_" + subtaskId; + return getDbzProperties().getProperty(SLOT_NAME.name()) + "_" + getSubtaskId(); } @Override diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java index 1490dd400..fb9054bd4 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java @@ -50,6 +50,8 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory { private List<String> schemaList; + private int lsnCommitCheckpointsDelay; + /** Creates a new {@link PostgresSourceConfig} for the given subtask {@code subtaskId}. */ @Override public PostgresSourceConfig create(int subtaskId) { @@ -100,7 +102,7 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory { // The PostgresSource will do snapshot according to its StartupMode. // Do not need debezium to do the snapshot work. - props.put("snapshot.mode", "never"); + props.setProperty("snapshot.mode", "never"); Configuration dbzConfiguration = Configuration.from(props); return new PostgresSourceConfig( @@ -129,7 +131,8 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory { connectionPoolSize, chunkKeyColumn, skipSnapshotBackfill, - scanNewlyAddedTableEnabled); + scanNewlyAddedTableEnabled, + lsnCommitCheckpointsDelay); } /** @@ -173,4 +176,9 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory { public void heartbeatInterval(Duration heartbeatInterval) { this.heartbeatInterval = heartbeatInterval; } + + /** The lsn commit checkpoints delay for Postgres. */ + public void setLsnCommitCheckpointsDelay(int lsnCommitCheckpointsDelay) { + this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay; + } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java index 391458ded..0443fa2e1 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java @@ -78,4 +78,14 @@ public class PostgresSourceOptions extends JdbcSourceOptions { .defaultValue(Duration.ofSeconds(30)) .withDescription( "Optional interval of sending heartbeat event for tracing the latest available replication slot offsets"); + + public static final ConfigOption<Integer> SCAN_LSN_COMMIT_CHECKPOINTS_DELAY = + ConfigOptions.key("scan.lsn-commit.checkpoints-num-delay") + .intType() + .defaultValue(3) + .withDescription( + "The number of checkpoint delays before starting to commit the LSN offsets.\n" + + "By setting this to higher value, the offset that is consumed by global slot will be " + + "committed after multiple checkpoint delays instead of after each checkpoint completion.\n" + + "This allows continuous recycle of log files in stream phase."); } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java index f6d201005..81bff8488 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReader.java @@ -28,6 +28,7 @@ import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitSeriali import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit; import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReaderContext; import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceReaderWithCommit; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig; import org.apache.flink.cdc.connectors.postgres.source.events.OffsetCommitAckEvent; import org.apache.flink.cdc.connectors.postgres.source.events.OffsetCommitEvent; import org.apache.flink.configuration.Configuration; @@ -38,6 +39,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; +import java.util.PriorityQueue; import java.util.function.Supplier; /** @@ -54,6 +56,9 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit { /** whether to commit offset. */ private volatile boolean isCommitOffset = false; + private final PriorityQueue<Long> minHeap; + private final int lsnCommitCheckpointsDelay; + public PostgresSourceReader( FutureCompletingBlockingQueue elementQueue, Supplier supplier, @@ -72,6 +77,9 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit { sourceConfig, sourceSplitSerializer, dialect); + this.lsnCommitCheckpointsDelay = + ((PostgresSourceConfig) sourceConfig).getLsnCommitCheckpointsDelay(); + this.minHeap = new PriorityQueue<>(); } @Override @@ -104,12 +112,23 @@ public class PostgresSourceReader extends IncrementalSourceReaderWithCommit { @Override public void notifyCheckpointComplete(long checkpointId) throws Exception { + this.minHeap.add(checkpointId); + if (this.minHeap.size() <= this.lsnCommitCheckpointsDelay) { + LOG.info("Pending checkpoints '{}'.", this.minHeap); + return; + } + final long checkpointIdToCommit = this.minHeap.poll(); + LOG.info( + "Pending checkpoints '{}', to be committed checkpoint id '{}'.", + this.minHeap, + checkpointIdToCommit); + // After all snapshot splits are finished, update stream split's metadata and reset start // offset, which maybe smaller than before. // In case that new start-offset of stream split has been recycled, don't commit offset // during new table added phase. if (isCommitOffset()) { - super.notifyCheckpointComplete(checkpointId); + super.notifyCheckpointComplete(checkpointIdToCommit); } } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java index e0e1b6b9b..5674908ae 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactory.java @@ -57,6 +57,7 @@ import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSou import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_ENABLED; +import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_LSN_COMMIT_CHECKPOINTS_DELAY; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_STARTUP_MODE; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SLOT_NAME; @@ -114,6 +115,7 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory { boolean closeIdlerReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); boolean isScanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED); + int lsnCommitCheckpointsDelay = config.get(SCAN_LSN_COMMIT_CHECKPOINTS_DELAY); if (enableParallelRead) { validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); @@ -158,7 +160,8 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory { chunkKeyColumn, closeIdlerReaders, skipSnapshotBackfill, - isScanNewlyAddedTableEnabled); + isScanNewlyAddedTableEnabled, + lsnCommitCheckpointsDelay); } @Override @@ -200,6 +203,7 @@ public class PostgreSQLTableFactory implements DynamicTableSourceFactory { options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED); + options.add(SCAN_LSN_COMMIT_CHECKPOINTS_DELAY); return options; } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java index add3d8f3d..9a9dd822d 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableSource.java @@ -82,10 +82,9 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe private final StartupOptions startupOptions; private final String chunkKeyColumn; private final boolean closeIdleReaders; - private final boolean skipSnapshotBackfill; - private final boolean scanNewlyAddedTableEnabled; + private final int lsnCommitCheckpointsDelay; // -------------------------------------------------------------------------------------------- // Mutable attributes @@ -124,7 +123,8 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe @Nullable String chunkKeyColumn, boolean closeIdleReaders, boolean skipSnapshotBackfill, - boolean isScanNewlyAddedTableEnabled) { + boolean isScanNewlyAddedTableEnabled, + int lsnCommitCheckpointsDelay) { this.physicalSchema = physicalSchema; this.port = port; this.hostname = checkNotNull(hostname); @@ -155,6 +155,7 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe this.closeIdleReaders = closeIdleReaders; this.skipSnapshotBackfill = skipSnapshotBackfill; this.scanNewlyAddedTableEnabled = isScanNewlyAddedTableEnabled; + this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay; } @Override @@ -216,6 +217,7 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe .closeIdleReaders(closeIdleReaders) .skipSnapshotBackfill(skipSnapshotBackfill) .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled) + .lsnCommitCheckpointsDelay(lsnCommitCheckpointsDelay) .build(); return SourceProvider.of(parallelSource); } else { @@ -283,7 +285,8 @@ public class PostgreSQLTableSource implements ScanTableSource, SupportsReadingMe chunkKeyColumn, closeIdleReaders, skipSnapshotBackfill, - scanNewlyAddedTableEnabled); + scanNewlyAddedTableEnabled, + lsnCommitCheckpointsDelay); source.metadataKeys = metadataKeys; source.producedDataType = producedDataType; return source; diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java index 10d64f226..5cc8ba5c5 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/PostgresTestBase.java @@ -220,6 +220,7 @@ public abstract class PostgresTestBase extends AbstractTestBase { postgresSourceConfigFactory.tableList(schemaName + "." + tableName); postgresSourceConfigFactory.splitSize(splitSize); postgresSourceConfigFactory.skipSnapshotBackfill(skipSnapshotBackfill); + postgresSourceConfigFactory.setLsnCommitCheckpointsDelay(1); return postgresSourceConfigFactory; } diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java index 96fa103f9..8d1a6d975 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java @@ -707,6 +707,7 @@ public class PostgresSourceITCase extends PostgresTestBase { .tableList(tableId) .startupOptions(startupOptions) .skipSnapshotBackfill(skipSnapshotBackfill) + .lsnCommitCheckpointsDelay(1) .deserializer(customerTable.getDeserializer()) .build(); @@ -816,7 +817,8 @@ public class PostgresSourceITCase extends PostgresTestBase { + " 'scan.startup.mode' = '%s'," + " 'scan.incremental.snapshot.chunk.size' = '100'," + " 'slot.name' = '%s'," - + " 'scan.incremental.snapshot.backfill.skip' = '%s'" + + " 'scan.incremental.snapshot.backfill.skip' = '%s'," + + " 'scan.lsn-commit.checkpoints-num-delay' = '1'" + "" + ")", customDatabase.getHost(), diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java new file mode 100644 index 000000000..1c855cc3f --- /dev/null +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceReaderTest.java @@ -0,0 +1,94 @@ +/* + * 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.reader; + +import org.apache.flink.cdc.connectors.postgres.source.MockPostgresDialect; +import org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder; +import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory; +import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory; +import org.apache.flink.cdc.connectors.postgres.testutils.TestTable; +import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema; +import org.apache.flink.connector.testutils.source.reader.TestingReaderContext; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ResolvedSchema; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.apache.flink.table.api.DataTypes.BIGINT; +import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link PostgresSourceReader}. */ +class PostgresSourceReaderTest { + + @Test + void testNotifyCheckpointWindowSizeOne() throws Exception { + final PostgresSourceReader reader = createReader(1); + final List<Long> completedCheckpointIds = new ArrayList<>(); + MockPostgresDialect.setNotifyCheckpointCompleteCallback( + id -> completedCheckpointIds.add(id)); + reader.notifyCheckpointComplete(11L); + assertThat(completedCheckpointIds).isEmpty(); + reader.notifyCheckpointComplete(12L); + assertThat(completedCheckpointIds).containsExactly(11L); + reader.notifyCheckpointComplete(13L); + assertThat(completedCheckpointIds).containsExactly(11L, 12L); + } + + @Test + void testNotifyCheckpointWindowSizeDefault() throws Exception { + final PostgresSourceReader reader = createReader(3); + final List<Long> completedCheckpointIds = new ArrayList<>(); + MockPostgresDialect.setNotifyCheckpointCompleteCallback( + id -> completedCheckpointIds.add(id)); + reader.notifyCheckpointComplete(103L); + assertThat(completedCheckpointIds).isEmpty(); + reader.notifyCheckpointComplete(102L); + assertThat(completedCheckpointIds).isEmpty(); + reader.notifyCheckpointComplete(101L); + assertThat(completedCheckpointIds).isEmpty(); + reader.notifyCheckpointComplete(104L); + assertThat(completedCheckpointIds).containsExactly(101L); + } + + private PostgresSourceReader createReader(final int lsnCommitCheckpointsDelay) + throws Exception { + final PostgresOffsetFactory offsetFactory = new PostgresOffsetFactory(); + final PostgresSourceConfigFactory configFactory = new PostgresSourceConfigFactory(); + configFactory.hostname("host"); + configFactory.database("pgdb"); + configFactory.username("username"); + configFactory.password("password"); + configFactory.setLsnCommitCheckpointsDelay(lsnCommitCheckpointsDelay); + final TestTable customerTable = + new TestTable( + "pgdb", + "customer", + "customers", + ResolvedSchema.of(Column.physical("id", BIGINT()))); + final DebeziumDeserializationSchema<?> deserializer = customerTable.getDeserializer(); + MockPostgresDialect dialect = new MockPostgresDialect(configFactory.create(0)); + final PostgresSourceBuilder.PostgresIncrementalSource<?> source = + new PostgresSourceBuilder.PostgresIncrementalSource<>( + configFactory, checkNotNull(deserializer), offsetFactory, dialect); + return source.createReader(new TestingReaderContext()); + } +} diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java index 9fe55b18f..199ff24fe 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/MockPostgreSQLTableSource.java @@ -64,7 +64,8 @@ public class MockPostgreSQLTableSource extends PostgreSQLTableSource { (String) get(postgreSQLTableSource, "chunkKeyColumn"), (boolean) get(postgreSQLTableSource, "closeIdleReaders"), (boolean) get(postgreSQLTableSource, "skipSnapshotBackfill"), - (boolean) get(postgreSQLTableSource, "scanNewlyAddedTableEnabled")); + (boolean) get(postgreSQLTableSource, "scanNewlyAddedTableEnabled"), + (int) get(postgreSQLTableSource, "lsnCommitCheckpointsDelay")); } @Override diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java index eb7360689..50cd9b7a4 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLTableFactoryTest.java @@ -59,6 +59,7 @@ import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSou import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.CONNECT_TIMEOUT; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.HEARTBEAT_INTERVAL; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; +import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_LSN_COMMIT_CHECKPOINTS_DELAY; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; import static org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; @@ -151,7 +152,8 @@ public class PostgreSQLTableFactoryTest { null, SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT, SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(), - SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue()); + SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(), + SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue()); assertEquals(expectedSource, actualSource); } @@ -196,7 +198,8 @@ public class PostgreSQLTableFactoryTest { null, SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT, true, - true); + true, + SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue()); assertEquals(expectedSource, actualSource); } @@ -239,7 +242,8 @@ public class PostgreSQLTableFactoryTest { null, SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT, SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(), - SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue()); + SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(), + SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue()); expectedSource.producedDataType = SCHEMA_WITH_METADATA.toSourceRowDataType(); expectedSource.metadataKeys = Arrays.asList("op_ts", "database_name", "schema_name", "table_name"); @@ -292,7 +296,8 @@ public class PostgreSQLTableFactoryTest { null, SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT, SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(), - SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue()); + SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(), + SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue()); assertEquals(expectedSource, actualSource); } @@ -335,7 +340,8 @@ public class PostgreSQLTableFactoryTest { null, SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED_DEFAULT, SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP.defaultValue(), - SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue()); + SCAN_NEWLY_ADDED_TABLE_ENABLED.defaultValue(), + SCAN_LSN_COMMIT_CHECKPOINTS_DELAY.defaultValue()); assertEquals(expectedSource, actualSource); }