[FLINK-40623][cdc-base] Release chunk splitter JDBC connection once a table is split

[xucq07]

What is the purpose of this pull request?

Fixes 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.

Since FLINK-34688 (asynchronous chunk splitting), 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#generateSplits borrows a pooled connection with try-with-resources and returns it when the call ends. analyzeTable and splitOneUnevenlySizedChunk take that connection as a parameter.
• The splitter no longer keeps a connection field, so open() and close() have nothing to acquire or release, and a splitter that has nothing left to split (e.g. restored after the snapshot phase) holds no connection.
• For unevenly sized tables, which are split one chunk per generateSplits call, the connection is now borrowed and returned per chunk instead of being held until the job ends.
• Added JdbcSourceChunkSplitterTest#testConnectionIsReleasedWhenEvenlySizedTableIsSplit and #testConnectionIsReleasedAfterEachUnevenlySizedChunk.

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 tests in flink-cdc-base:
  • testConnectionIsReleasedWhenEvenlySizedTableIsSplit asserts that open() does not open a connection, and that splitting an evenly sized table opens and closes exactly one connection.
  • testConnectionIsReleasedAfterEachUnevenlySizedChunk splits an unevenly sized table in two calls and asserts that each call opens and closes its own connection, so none is held between calls.
• 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 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.

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?

• Yes (Claude Code, Claude Fable 5.1)

Generated-by: Claude Code (Claude Fable 5.1)

🤖 Generated with Claude Code

[loserwang1024]

What about using temp jdbc connection each time for generateSplits and release immediately. code is more clean.

Suggested change

	public Collection<SnapshotSplit> generateSplits(TableId tableId) throws Exception {
	@Override
	public Collection<SnapshotSplit> generateSplits(TableId tableId) throws Exception {
	try (JdbcConnection jdbcConnection = dialect.openJdbcConnection(sourceConfig)){
	return generateSplits(tableId, jdbcConnection);
	}
	}
	private Collection<SnapshotSplit> generateSplits(TableId tableId, JdbcConnection jdbcConnection) throws Exception {
	if (!hasNextChunk()) {
	// split a new table.
	analyzeTable(tableId, jdbcConnection);
	Optional<List<SnapshotSplit>> evenlySplitChunks = trySplitAllEvenlySizedChunks(tableId);
	if (evenlySplitChunks.isPresent()) {
	return evenlySplitChunks.get();
	} else {
	synchronized (lock) {
	this.currentSplittingTableId = tableId;
	this.nextChunkStart = ChunkSplitterState.ChunkBound.START_BOUND;
	this.nextChunkId = 0;
	return Collections.singletonList(splitOneUnevenlySizedChunk(tableId, jdbcConnection));
	}
	}
	} else {
	Preconditions.checkState(
	currentSplittingTableId.equals(tableId),
	"Can not split a new table before the previous table splitting finish.");
	if (currentSplittingTable == null) {
	analyzeTable(currentSplittingTableId, jdbcConnection);
	}
	synchronized (lock) {
	return Collections.singletonList(splitOneUnevenlySizedChunk(tableId, jdbcConnection));
	}
	}
	}

[xucq07]

Done, thanks for the suggestion.

[loserwang1024]

LGTM