Hi Flink CDC devs,
I'd like to report a bug in the CDC incremental source framework
(flink-cdc-base) that I hit while upgrading our PostgreSQL CDC jobs
from 3.2.1 to 3.6.0. I have requested an ASF JIRA account but it is
not approved yet, so I'm posting the full analysis here first. If a
committer prefers to open the ticket on my behalf, please feel free
to; otherwise I will file it as soon as the account is approved
(component: Flink CDC).
== Summary ==
Since FLINK-34688 (asynchronous chunk splitting, PR #3510, included
from cdc-3.3.0 on), JdbcSourceChunkSplitter opens its JDBC connection
in open() and only closes it in close(). SnapshotSplitAssigner calls
chunkSplitter.close() only from its own close(), i.e. when the job
terminates, not when splitting is finished. The connection comes from
the HikariCP pool in JdbcConnectionPools, which is a JVM-wide
singleton shared by every job on the same JobManager that targets the
same host/port/user/database. The pool size is connection.pool.size
(default 20) and the borrow timeout is connect.timeout (default 30 s).
Net effect: every running incremental-snapshot job pins one pooled
connection on the JobManager for its whole lifetime. Once 20 such jobs
are running against the same database, the 21st job fails in
createEnumerator -> discoverDataCollections() with a Hikari timeout
and loops in RESTARTING forever. Nothing in the message points at
connection.pool.size.
== Environment ==
* Flink 1.20.0 standalone session cluster (1 JM + 1 TM), JDK 17
* flink-sql-connector-postgres-cdc 3.6.0-1.20 (Debezium 1.9.8.Final)
* PostgreSQL 15 (cloud RDS), 30 independent Flink SQL jobs, one
postgres-cdc table each, all on the same host/port/user/database
* scan.incremental.snapshot.enabled = true, scan.startup.mode =
initial, connection.pool.size not set (default 20)
* 3.2.1 with the same DDL runs all 30 jobs fine; 3.2.1 still used
try-with-resources per generateSplits().
== Symptom ==
Jobs 1..20 start normally. From job 21 on, every new job fails during
enumerator creation:
Caused by: org.apache.flink.util.FlinkRuntimeException: Failed to
discover captured tables for enumerator
Caused by: org.apache.flink.util.FlinkRuntimeException:
java.sql.SQLTransientConnectionException: connection-pool-<host>:5432
- Connection is not available, request timed out after 30000ms.
at
org.apache.flink.cdc.connectors.base.relational.connection.JdbcConnectionFactory.connect(...)
at
org.apache.flink.cdc.connectors.postgres.source.PostgresDialect.openJdbcConnection(...)
at
org.apache.flink.cdc.connectors.postgres.source.PostgresDialect.discoverDataCollections(...)
at
org.apache.flink.cdc.connectors.postgres.source.PostgresSourceBuilder$PostgresIncrementalSource.createEnumerator(...)
* Submitting the failing job alone (after cancelling the other failing
ones) does not help.
* Restarting the TaskManager does not help (the pool that is exhausted
lives in the JobManager).
* Restarting the JobManager "fixes" it until 20 jobs are running again.
* pg_stat_activity shows exactly 20 idle backends from the JobManager,
one per running job, created at each job's start and never used again.
Their last statement is the chunk-split query:
SELECT MAX("id") FROM (SELECT "id" FROM "<db>"."public"."<table>"
WHERE "id" >= $1 ORDER BY "id" ASC LIMIT 8096) AS T
== Code references (release-3.6) ==
* flink-cdc-base .../source/assigner/splitter/JdbcSourceChunkSplitter.java
open(): this.jdbcConnection = dialect.openJdbcConnection(sourceConfig);
close(): if (jdbcConnection != null) { jdbcConnection.close(); }
release-3.2 had: try (JdbcConnection jdbc =
dialect.openJdbcConnection(sourceConfig)) { ... } inside
generateSplits().
* flink-cdc-base .../source/assigner/SnapshotSplitAssigner.java
open() -> chunkSplitter.open(); ... startAsynchronouslySplit();
close() -> chunkSplitter.close(); // the only place it is closed
onFinishedSplits() flips assignerStatus to finished but does not
close the splitter.
* flink-cdc-base .../relational/connection/JdbcConnectionFactory.java
JdbcConnectionPools.getInstance(factory).getOrCreateConnectionPool(poolId,
sourceConfig).getConnection()
JdbcConnectionPools is a singleton; the pool is keyed per
host/port/user/database, i.e. shared across jobs in the JVM.
* flink-cdc-base .../relational/connection/JdbcConnectionPoolFactory.java
config.setMaximumPoolSize(sourceConfig.getConnectionPoolSize());
// default 20
config.setConnectionTimeout(sourceConfig.getConnectTimeout().toMillis());
// default 30 s
== Expected behaviour / proposal ==
The chunk splitter should give its connection back once splitting for
all remaining tables is done (e.g. close it at the end of
splitChunksForRemainingTables(), or go back to a try-with-resources
per generateSplits() call, re-opening lazily if newly added tables
arrive). Then the number of concurrently running jobs is no longer
bounded by connection.pool.size on the JobManager. At minimum the
limitation should be documented and the timeout should be reported
with a hint about connection.pool.size.
== Workaround ==
Set 'connection.pool.size' in the table DDL to a value larger than the
number of jobs sharing the JobManager (we use 64 for 30 jobs) and
restart the JobManager, because the pool size is fixed when the pool
is first created.
== How to reproduce ==
1. Session cluster, postgres-cdc 3.6.0, incremental snapshot enabled,
connection.pool.size at default.
2. Submit 21 independent SQL jobs, each reading one table from the
same PostgreSQL database.
3. Job 21 fails in createEnumerator with the exception above;
pg_stat_activity shows 20 idle JobManager backends whose last query is
the chunk-split SELECT MAX(...) LIMIT 8096.
I'm happy to work on a fix once the JIRA exists and is assigned.
Thanks.
