wglbd12 created FLINK-40623:
-------------------------------
Summary: [cdc-base] JdbcSourceChunkSplitter keeps a pooled JDBC
connection for the whole job lifetime; with default connection.pool.size=20 the
21st incremental-snapshot job on the same JobManager can never start
Key: FLINK-40623
URL: https://issues.apache.org/jira/browse/FLINK-40623
Project: Flink
Issue Type: Bug
Components: Flink CDC
Affects Versions: cdc-3.6.0
Reporter: wglbd12
h3. 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)
* Same DDL on 3.2.1 runs all 30 jobs without issue
h3. Symptom
Jobs 1..20 start normally. From job 21 on every new job fails during enumerator
creation and loops in RESTARTING:
{code}
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(...)
{code}
* Submitting the failing job alone does not help; restarting the TaskManager
does not help; 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:
{code}
SELECT MAX("id") FROM (SELECT "id" FROM "<db>"."public"."<table>" WHERE "id" >=
$1 ORDER BY "id" ASC LIMIT 8096) AS T
{code}
h3. Root cause (release-3.6 sources)
* {{JdbcSourceChunkSplitter}}: {{open()}} does {{this.jdbcConnection =
dialect.openJdbcConnection(sourceConfig)}}, {{close()}} closes it. In
release-3.2 {{generateSplits()}} used {{try (JdbcConnection jdbc =
dialect.openJdbcConnection(sourceConfig))}}, so the connection was returned
after each table. The lifecycle change came with FLINK-34688 (asynchronous
chunk splitting, PR #3510), first shipped in cdc-3.3.0.
* {{SnapshotSplitAssigner}}: {{open()}} calls {{chunkSplitter.open()}};
{{chunkSplitter.close()}} is only called from the assigner's own {{close()}},
i.e. at job termination. {{onFinishedSplits()}} flips the assigner status to
finished but does not release the splitter's connection.
* {{JdbcConnectionFactory.connect()}} borrows from
{{JdbcConnectionPools.getInstance(...).getOrCreateConnectionPool(poolId,
sourceConfig)}}; {{JdbcConnectionPools}} is a JVM-wide singleton keyed per
host/port/user/database, therefore shared by all jobs on the same JobManager.
* {{JdbcConnectionPoolFactory.createPooledDataSource()}}:
{{setMaximumPoolSize(sourceConfig.getConnectionPoolSize())}} (default 20) and
{{setConnectionTimeout(sourceConfig.getConnectTimeout().toMillis())}} (default
30 s).
Net effect: each running incremental-snapshot job pins one pooled connection on
the JobManager for its whole lifetime, so the number of jobs that can run
against one database from one JobManager is capped at {{connection.pool.size}},
and the failure surfaces as an opaque Hikari timeout in
{{discoverDataCollections()}}.
h3. Expected behaviour
The chunk splitter should release its JDBC connection once splitting for all
remaining tables has finished (e.g. at the end of
{{splitChunksForRemainingTables()}}, re-opening lazily for newly added tables),
or return to a try-with-resources per {{generateSplits()}}. At minimum,
document the limitation and make the error mention {{connection.pool.size}}.
h3. Workaround
Set {{'connection.pool.size'}} in the table DDL to a value larger than the
number of jobs sharing the JobManager (64 for our 30 jobs) and restart the
JobManager; the pool size is fixed when the pool is first created.
h3. How to reproduce
# Session cluster, postgres-cdc 3.6.0, incremental snapshot enabled,
{{connection.pool.size}} at default.
# Submit 21 independent SQL jobs, each reading one table from the same
PostgreSQL database.
# 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}}.
Mailing list thread with the same analysis:
https://lists.apache.org/thread/0vs1v0yl2t63qst0fqopy0fdvxr1qq8g
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
