[
https://issues.apache.org/jira/browse/FLINK-40623?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40623:
-----------------------------------
Labels: pull-request-available (was: )
> [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
> 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)
