xucq07 opened a new pull request, #4534: URL: https://github.com/apache/flink-cdc/pull/4534
## What is the purpose of this pull request? Fixes [FLINK-40623](https://issues.apache.org/jira/browse/FLINK-40623): every incremental-snapshot JDBC source (Postgres, Oracle, SQL Server, Db2) keeps one JDBC connection open on the JobManager for the whole lifetime of the job. `JdbcSourceChunkSplitter#open()` acquires a connection from `JdbcConnectionPools` and only `close()` releases it. The splitter is only needed while a table is being split into chunks, but the enumerator keeps it alive until the job ends, so the connection stays checked out during the entire streaming phase. `JdbcConnectionPools` is a JVM-wide singleton keyed by host/port/user/database with a default `connection.pool.size` of 20, so on a session cluster the 21st job against the same database fails at submission with: ``` java.sql.SQLTransientConnectionException: connection-pool-...-1 - Connection is not available, request timed out after 30000ms. ``` This is not fixable from the job side: `connection.pool.size` is only honoured by whichever job creates the pool first, and `debezium.connect.pool.size` is not read by the pool at all. ## Brief change log - `JdbcSourceChunkSplitter` acquires its JDBC connection lazily on first use and releases it as soon as the current table has no more chunks (`hasNextChunk() == false`) or the splitter is closed. `open()` no longer touches the pool, so an assigner restored after the snapshot phase never holds a connection. - `releaseConnection()` is non-throwing and idempotent, so a close failure in the `finally` block cannot mask the original exception from `generateSplits`. - Added `JdbcSourceChunkSplitterTest#testConnectionIsAcquiredLazilyAndReleasedWhenTableIsSplit`. Other chunk splitters (`MySqlChunkSplitter`, MongoDB) do not extend `JdbcSourceChunkSplitter` and are unaffected. ## Verifying this change This change added tests and can be verified as follows: - Added unit test in `flink-cdc-base`: `JdbcSourceChunkSplitterTest#testConnectionIsAcquiredLazilyAndReleasedWhenTableIsSplit` asserts that `open()` does not open a connection, that exactly one connection is opened and closed while splitting a single-chunk table, and that `close()` is idempotent. - Full CI matrix (all source/pipeline unit and e2e jobs, including the `postgres-source` groups on Flink 1.20.3 and 2.2.0) passed on my fork: https://github.com/xucq07/flink-cdc/actions/runs/34737364170 - Manually tested on a Flink 1.20.0 standalone session cluster with 30 Flink SQL `postgres-cdc` jobs (incremental snapshot enabled) against one PostgreSQL database, default `connection.pool.size` (20): - Before (3.6.0): JobManager held 31 connections to the database, one per job; submitting jobs beyond the pool size fails with the timeout above. - After (this patch built as `flink-sql-connector-postgres-cdc`): all 30 jobs submitted from a fresh snapshot, JobManager connections dropped to 1 to 2 once splitting finished, TaskManager connections unchanged, no `HikariPool` errors, checkpoints completing for all jobs, and a row updated in the source table arrived in the sink within seconds. Restoring 15 jobs from savepoints also left the JobManager with no splitter connection. ## Documentation - Does this pull request introduce a new feature? (no) - If yes, how is the feature documented? (not applicable) --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (Claude Code, Claude Fable 5.1) Generated-by: Claude Code (Claude Fable 5.1) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
