[Feature][Connector-V2][PostgreSQL-CDC] Add committed-offset and snapshot-only startup modes

[CloverDew]

Purpose of this pull request

Close: #11039

This PR adds two PostgreSQL CDC startup modes:

• committed-offset: starts WAL streaming from the configured replication slot's committed LSN and skips snapshot reading.
• snapshot-only: reads the startup snapshot and finishes as a bounded job without entering WAL streaming.

Does this PR introduce any user-facing change?

Yes.
PostgreSQL CDC now accepts two additional startup.mode values:

• snapshot-only
  • Previous behavior: PostgreSQL CDC snapshot startup always transitioned into WAL streaming.
  • New behavior: the job reads snapshot data and then finishes naturally as a bounded job.
• committed-offset
  • Previous behavior: PostgreSQL CDC supported initial, earliest, and latest, but could not explicitly start from the committed position of an existing logical replication slot.
  • New behavior: the job starts from the configured replication slot's confirmed_flush_lsn, skips snapshot data, and fails fast if slot.name is missing, the slot does not exist, or the slot has no usable committed LSN.

How was this patch tested?

Added PostgreSQL CDC E2E coverage for the new startup modes:

• snapshot-only verifies that snapshot data is written and later source changes are not streamed.
• committed-offset verifies that the connector attaches to an existing logical replication slot and does not snapshot existing rows.

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[DanielLeens]

Thanks for the contribution! I re-reviewed the latest head f714fc533e89 from scratch and traced the new startup modes through the real PostgreSQL CDC code path. The implementation looks solid to me overall, and I did not find a new source-side blocker in the current version.

What this PR fixes

• User pain: PostgreSQL-CDC was still missing two practical startup modes that people often need operationally: a bounded snapshot-only run, and a slot-based startup from an already committed replication position.
• Fix approach: extend the PostgreSQL CDC startup-mode surface with snapshot-only and committed-offset, wire them into startup offset resolution / boundedness / split assignment, and add E2E + docs coverage.
• One-line summary: the new modes are implemented as explicit SeaTunnel behaviors rather than hidden Debezium passthrough, and the main source/enumerator/reader path is aligned with the documented semantics.

Full call path I checked

snapshot-only
  -> startup.mode = snapshot-only
  -> PostgresSourceOptions.STARTUP_MODE accepts SNAPSHOT_ONLY
  -> StartupConfig.getStartupOffset() returns null
  -> IncrementalSource.getBoundedness() => BOUNDED
  -> IncrementalSource.createEnumerator()
      -> SnapshotOnlySplitAssigner
      -> snapshot splits only
  -> PostgresSourceFetchTaskContext.configure()
      -> snapshotter does not enter WAL streaming
  -> job finishes naturally after the snapshot phase

committed-offset
  -> startup.mode = committed-offset
  -> PostgresIncrementalSource.validateStartupOptions()
      -> require explicit slot.name
  -> StartupConfig.getStartupOffset()
      -> offsetFactory.committedOffset()
  -> LsnOffsetFactory.committedOffset()
      -> SELECT confirmed_flush_lsn::text FROM pg_replication_slots WHERE slot_name = ?
      -> build LsnOffset from the slot's committed LSN
  -> IncrementalSplitAssigner.createIncrementalSplit()
      -> use that LSN as the incremental split startup offset
  -> PostgresSourceFetchTaskContext.loadStartingOffsetState()
      -> rebuild PostgresOffsetContext from the startup offset
  -> PostgresWalFetchTask.execute()
      -> Debezium streaming starts from that WAL position

Findings

I did not find a new blocker in the current version.

The key points I explicitly cross-checked were:

• snapshot-only really becomes a bounded source (IncrementalSource.getBoundedness() + SnapshotOnlySplitAssigner)
• committed-offset really comes from the replication slot's committed LSN instead of a snapshot-side workaround
• the fetch context no longer tries to create a slot when the slot already exists and the job is supposed to reuse it
• both EN and ZH docs were updated to describe the new modes consistently

Test coverage / stability note

For the test-stability pass, I would rate the new PostgreSQL CDC E2E coverage as stable. The new test structure does not depend on fixed ports, shared mutable static test state, or timing-only assertions for job completion. The replication-slot setup/cleanup is also handled deliberately, which reduces cross-test contamination in the shared Postgres container.

One small non-blocking follow-up I would still consider: the committed-offset E2E currently proves that existing rows are not snapshotted again, which is good, but it could be even stronger if it also inserted one new row after the job starts and asserted that the new WAL event reaches the sink. That would prove both halves of the contract in one place.

Review conclusion

Conclusion: can merge

1. Blocking items

• None from my side.

2. Non-blocking suggestions

• Consider extending the committed-offset E2E with one post-start insert so the test verifies both "skip historical rows" and "continue streaming new WAL changes".

Overall this is a clean, focused PostgreSQL-CDC feature addition. The startup-mode parsing, offset path, boundedness behavior, docs, and E2E coverage are aligned well enough for me to approve.

[DanielLeens]

Rechecked the latest head: my previous Daniel-side conclusion had no remaining source blocker and was waiting only on CI. The previously blocking Build check is now successful on this head, so I am updating my review to approve.

I also recorded the base/head sync fact for the queue. This branch is still diverged with behind_by=11, but because the current required Build signal is green, I am not asking for a sync-only retry from Daniel's side here. If GitHub's merge gate later asks for an update branch step, please handle that as a merge-gate action.

[DanielLeens]

Rechecked the latest head: the source-level blocker from my previous Daniel round is still closed for me, and the current Build gate is now green on this unchanged head.

At this point I do not see a new code-side blocker from Daniel on this revision. For queue bookkeeping, this branch is still diverged against the latest dev (ahead_by=2, behind_by=19), but because the required Build signal is green now, I am not asking for a sync-only retry from Daniel's side here.

If the repository still needs a write-capable maintainer to refresh the formal review state or dismiss an older visible changes requested entry, that final gate may need to be handled separately from Daniel's source-level recheck.

[DanielLeens]

Sorry for missing these points in my earlier reviews - that's on me. I re-reviewed the full latest head f714fc533e89 from scratch, including the SeaTunnel enumerator/reader lifecycle, Debezium 1.9.8 snapshotter behavior, replication-slot ownership, and the updated E2E. My earlier approval and today's source-cleared comment are not the right technical conclusion for this revision.

Thank you for taking on these two useful PostgreSQL CDC startup modes. The overall direction is good: snapshot-only is bounded at the SeaTunnel enumerator layer, and committed-offset resolves the slot's confirmed_flush_lsn. However, I found two source-level blockers that need to be closed before merging.

Full runtime path checked

snapshot-only
  -> PostgresSourceOptions.STARTUP_MODE
  -> IncrementalSource.getBoundedness() returns BOUNDED
  -> SnapshotOnlySplitAssigner emits snapshot splits only
  -> reader receives a SnapshotSplit
  -> PostgresSourceConfigFactory.create()
      -> does not map SeaTunnel snapshot-only to Debezium snapshot.mode=initial_only
      -> Debezium defaults to snapshot.mode=initial
  -> PostgresSourceFetchTaskContext.configure()
      -> InitialSnapshotter.shouldStream() returns true
      -> creates a replication connection
      -> creates the logical replication slot when it does not exist
  -> snapshot finishes and the enumerator signals NoMoreSplits
  -> bounded job exits
  -> slot.drop.on.stop defaults to false
  -> the slot remains without a WAL consumer advancing confirmed_flush_lsn

committed-offset
  -> validates that slot.name is explicitly configured
  -> LsnOffsetFactory.committedOffset()
      -> reads pg_replication_slots.confirmed_flush_lsn
      -> builds the Debezium PostgreSQL offset
  -> IncrementalSplitAssigner creates the incremental split
  -> PostgresSourceFetchTaskContext rebuilds PostgresOffsetContext
  -> PostgresWalFetchTask
  -> Debezium PostgresStreamingChangeEventSource
      -> starts from the committed WAL position
      -> locates the transaction boundary
      -> reads WAL events
      -> advances the committed offset after checkpoint completion

Blocking issue 1: snapshot-only still creates and retains a logical replication slot

• Location: seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java:58, and PostgresSourceFetchTaskContext.java:133,197-210
• What happens: the new mode stops incremental split assignment only at the SeaTunnel layer. The reader still uses Debezium's default snapshot.mode=initial; InitialSnapshotter.shouldStream() is true, so configure() enters replication setup and creates the slot if it is absent.
• Risk: after the bounded job exits, the slot is not dropped by default and no reader advances it. PostgreSQL can keep retaining WAL for that slot until disk usage becomes operationally dangerous. It can also collide with the default slot name and unnecessarily require replication privileges for a snapshot-only job.
• Recommended fix: preferably make PostgresSourceFetchTaskContext explicitly skip replication setup for StartupMode.SNAPSHOT_ONLY. Alternatively, force Debezium snapshot.mode=initial_only as an internal contract after user Debezium properties are applied. Please add a regression assertion proving that the job creates no active or retained slot after completion.

Blocking issue 2: the latest E2E can pass even when committed-offset WAL streaming is broken

• Location: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java:419,449-475
• What happens: the latest fix ci revision removed the earlier positive WAL row/assertion. The test now checks only that historical rows 10 and 11 are absent. untilAsserted(isEmpty()) can succeed on its first empty observation, and the CompletableFuture is neither retained nor joined, so a background job failure does not fail the test.
• Risk: CI stays green even if offset reconstruction, WAL startup, incremental delivery, or the async job itself is broken. The new SnapshotPhaseState restore branch is also not exercised by checkpoint/savepoint recovery.
• Recommended fix: create the slot and committed position, write at least one event after that position, start the job, then assert post-start insert/update/delete delivery. Please retain and join the future so async failures propagate. Also add focused slot-missing/null-LSN coverage and a checkpoint/restore test for the new snapshot-only state path.

Test stability rating: high risk, because the current committed-offset E2E has a false-positive path rather than merely a timing-related flaky risk.

Review conclusion

Conclusion: changes are required before merge

1. Blocking items

• Make snapshot-only avoid creating or retaining a PostgreSQL replication slot, and verify the external resource lifecycle.
• Restore a positive committed-offset WAL assertion, propagate async failures, and cover the newly introduced checkpoint/restore path.

2. Non-blocking suggestions

• Replace the fixed five-second sleep in the snapshot-only E2E with a condition tied to job completion and slot state.
• Add a compatibility test for the shared StartupMode.toString() lowercase/hyphen behavior, or keep that conversion scoped to configuration parsing.

The current CI checks are green, which is helpful, but issue 2 is precisely why that green result does not clear the source blockers yet. The branch is also currently diverged from dev (ahead_by=2, behind_by=20); I am recording that as a merge-state fact, not asking for a sync-only retry, because it is not the cause of the two issues above.

The feature is worth continuing, and most of the structure is already in place. Once these two lifecycle and verification gaps are fixed, I will be happy to re-review the complete updated head promptly. Thanks again for the work and for your patience with the correction.