[
https://issues.apache.org/jira/browse/FLINK-40623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18114760#comment-18114760
]
wglbd12 commented on FLINK-40623:
---------------------------------
I have opened a PR with a fix for this issue:
https://github.com/apache/flink-cdc/pull/4534
The change makes JdbcSourceChunkSplitter acquire its JDBC connection lazily and
release it as soon as a table has been split, so the JobManager no longer holds
one pooled connection per job for the whole job lifetime. Verified with the
added unit test, the full CI matrix on my fork, and on a Flink 1.20 session
cluster with 30 postgres-cdc jobs against one database with the default
connection.pool.size (JobManager connections dropped from 31 to 1-2, all 30
jobs start).
Could a committer please assign this ticket to me? Thanks.
> [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)
