[jira] [Assigned] (FLINK-40623) [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

["Hongshun Wang (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40623?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Hongshun Wang reassigned FLINK-40623:
-------------------------------------

    Assignee: wglbd12

> [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
>            Assignee: wglbd12
>            Priority: Major
>              Labels: pull-request-available
>
> 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)