You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "tchivs (via GitHub)" <gi...@apache.org> on 2024/07/15 01:23:29 UTC
[I] [Bug] CheckPoint Timeout Due to Retrieving Schemas for All Subpartitions When Synchronizing a Partitioned PostgreSQL Table [doris-flink-connector]
tchivs opened a new issue, #426:
URL: https://github.com/apache/doris-flink-connector/issues/426
### Search before asking
- [X] I had searched in the [issues](https://github.com/apache/incubator-doris/issues?q=is%3Aissue) and found no similar issues.
### Version
1.6.1
### What's Wrong?
When synchronizing a PostgreSQL table using a connector, if the table contains a large number of partitions, the checkpoint always fails. By tracing the source code, it was found that the PostgresDialect's queryTableSchema queries the schema of each table. This schema querying during each checkpoint causes connection timeouts.
```shell
flink run \
-Dexecution.checkpointing.interval=10s \
-Dparallelism.default=1 \
-c com.hytera.cdc.FromPg2Doris \
./flink-doris-connector-1.19-1.6.1.jar\
--database aia_dev\
--postgres-conf scan.incremental.snapshot.enabled=true \
--postgres-conf hostname=xxxxx \
--postgres-conf port=5432 \
--postgres-conf username=postgres \
--postgres-conf password="xxxxx" \
--postgres-conf database-name=test \
--postgres-conf schema-name=public \
--postgres-conf slot.name=aia_test \
--postgres-conf decoding.plugin.name=pgoutput \
--table-prefix dwd_\
--ignore-default-value \
--including-tables "jjdb_.*|fkdb_.*|pjdb_.*|dsrdb|zjdb|case_log_test" \
--multi-to-one-origin "jjdb_.*|fkdb_.*|pjdb_.*" \
--multi-to-one-target "dwd_jjdb|dwdfkdb|dwd_pjdb" \
--excluding-tables ".*_bak"\
--sink-conf fenodes=10.115.6.215:8030 \
--sink-conf username=admin \
--sink-conf password=:a:74Y4nF-8m\
--sink-conf jdbc-url=jdbc:mysql://10.115.6.215:9030 \
--sink-conf sink.label-prefix=label \
--sink-conf sink.enable.batch-mode=false \
--sink-conf sink.properties.partial.columns=true\
--sink-conf sink.enable-delete=true\
--table-conf replication_num=3\
```
check point Exception:
```
org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerable failure threshold. The latest checkpoint failed due to Checkpoint expired before completing., view the Checkpoint History tab or the Job Manager log to find out why continuous checkpoints failed.
```
error log：
```logger
2024-05-28 15:39:07,367 INFO org.apache.flink.runtime.taskmanager.Task [] - Freeing task resources for jjdb: Writer -> jjdb: Committer (1/1)#0 (dbec67a546f03f19bae3a56726a02174_788573959fc6fd87fb6bfd0ffc27d896_0_0).
2024-05-28 15:39:07,370 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Un-registering task and sending final execution state CANCELED to JobManager for task fkdb: Writer -> fkdb: Committer (1/1)#0 dbec67a546f03f19bae3a56726a02174_1992c3287130e2f49268dbcc909a7d1d_0_0.
2024-05-28 15:39:07,370 INFO org.apache.http.impl.execchain.RetryExec [] - I/O exception (java.net.SocketException) caught when processing request to {}->http://192.168.0.168:8040: Socket closed
2024-05-28 15:39:07,378 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Un-registering task and sending final execution state CANCELED to JobManager for task jjdb: Writer -> jjdb: Committer (1/1)#0 dbec67a546f03f19bae3a56726a02174_788573959fc6fd87fb6bfd0ffc27d896_0_0.
2024-05-28 15:39:37,363 WARN org.apache.flink.runtime.taskmanager.Task [] - Task 'Source: Postgres Source -> Process (1/1)#0' did not react to cancelling signal - interrupting; it is stuck for 30 seconds in method:
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:858)
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:807)
org.apache.flink.runtime.taskmanager.Task$$Lambda$813/1723769838.run(Unknown Source)
org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:953)
org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932)
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:746)
org.apache.flink.runtime.taskmanager.Task.run(Task.java:562)
java.lang.Thread.run(Thread.java:750)
2024-05-28 15:39:37,372 INFO io.debezium.embedded.EmbeddedEngine [] - Stopping the embedded engine
2024-05-28 15:39:37,373 ERROR com.ververica.cdc.debezium.internal.Handover [] - Reporting error:
java.lang.InterruptedException: null
at java.lang.Object.wait(Native Method) ~[?:1.8.0_381]
at java.lang.Object.wait(Object.java:502) ~[?:1.8.0_381]
at com.ververica.cdc.debezium.internal.Handover.produce(Handover.java:115) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at com.ververica.cdc.debezium.internal.DebeziumChangeConsumer.handleBatch(DebeziumChangeConsumer.java:54) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.embedded.ConvertingEngineBuilder.lambda$notifying$2(ConvertingEngineBuilder.java:83) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:822) [flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:192) [flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_381]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_381]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_381]
2024-05-28 15:39:37,375 INFO io.debezium.embedded.EmbeddedEngine [] - Stopping the task and engine
2024-05-28 15:39:37,375 INFO io.debezium.connector.common.BaseSourceTask [] - Stopping down connector
2024-05-28 15:41:07,376 WARN io.debezium.pipeline.ChangeEventSourceCoordinator [] - Coordinator didn't stop in the expected time, shutting down executor now
2024-05-28 15:41:07,377 WARN io.debezium.pipeline.source.AbstractSnapshotChangeEventSource [] - Snapshot was interrupted before completion
2024-05-28 15:41:07,377 INFO io.debezium.pipeline.source.AbstractSnapshotChangeEventSource [] - Snapshot - Final stage
2024-05-28 15:41:07,377 WARN io.debezium.pipeline.ChangeEventSourceCoordinator [] - Change event source executor was interrupted
java.lang.InterruptedException: null
at java.lang.Object.wait(Native Method) ~[?:1.8.0_381]
at io.debezium.connector.base.ChangeEventQueue.doEnqueue(ChangeEventQueue.java:204) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.connector.base.ChangeEventQueue.enqueue(ChangeEventQueue.java:169) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.EventDispatcher$BufferingSnapshotChangeRecordReceiver.changeRecord(EventDispatcher.java:440) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.EventDispatcher$1.changeRecord(EventDispatcher.java:166) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.relational.RelationalChangeRecordEmitter.emitReadRecord(RelationalChangeRecordEmitter.java:120) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:57) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.EventDispatcher.dispatchSnapshotEvent(EventDispatcher.java:155) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.relational.RelationalSnapshotChangeEventSource.createDataEventsForTable(RelationalSnapshotChangeEventSource.java:407) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.relational.RelationalSnapshotChangeEventSource.createDataEvents(RelationalSnapshotChangeEventSource.java:316) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.relational.RelationalSnapshotChangeEventSource.doExecute(RelationalSnapshotChangeEventSource.java:132) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.source.AbstractSnapshotChangeEventSource.execute(AbstractSnapshotChangeEventSource.java:76) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.ChangeEventSourceCoordinator.doSnapshot(ChangeEventSourceCoordinator.java:155) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.ChangeEventSourceCoordinator.executeChangeEventSources(ChangeEventSourceCoordinator.java:137) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:109) ~[flink-sql-connector-postgres-cdc-3.0.1.jar:3.0.1]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_381]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_381]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_381]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_381]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_381]
2024-05-28 15:41:07,378 INFO io.debezium.pipeline.ChangeEventSourceCoordinator [] - Connected metrics set to 'false'
2024-05-28 15:41:07,380 INFO io.debezium.jdbc.JdbcConnection [] - Connection gracefully closed
```
### What You Expected?
When synchronizing a PostgreSQL table using a connector, if the table contains a large number of partitions, the checkpoint always fails. By tracing the source code, it was found that the `PostgresDialect`'s `queryTableSchema` queries the schema of each table. This schema querying during each checkpoint causes connection timeouts.
**Solution:** Use caching. Each partitioned table should match and fetch the schema only once using the following parameters:
- `--multi-to-one-origin "jjdb_.*|fkdb_.*|pjdb_.*"`
- `--multi-to-one-target "dwd_jjdb|dwdfkdb|dwd_pjdb"`
### How to Reproduce?
**Steps to Reproduce:**
1. Add a partitioned table in PostgreSQL.
2. Create partitions for nearly ten years.
3. Synchronize this table.
### Anything Else?
_No response_
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [I] [Bug] CheckPoint Timeout Due to Retrieving Schemas for All Subpartitions When Synchronizing a Partitioned PostgreSQL Table [doris-flink-connector]
Posted by "tchivs (via GitHub)" <gi...@apache.org>.
tchivs commented on issue #426:
URL: https://github.com/apache/doris-flink-connector/issues/426#issuecomment-2314130710
> 感谢您提出此问题。这似乎是 flink-cdc-connector 的问题。你可以尝试将你的修复贡献给 flink-cdc 社区。
OK，I have submitted issus to them, and this issue can be closed
[FLINK-36164](https://[issues.apache.org/jira/projects/FLINK/issues/FLINK-36164](https://issues.apache.org/jira/projects/FLINK/issues/FLINK-36164))
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [I] [Bug] CheckPoint Timeout Due to Retrieving Schemas for All Subpartitions When Synchronizing a Partitioned PostgreSQL Table [doris-flink-connector]
Posted by "tchivs (via GitHub)" <gi...@apache.org>.
tchivs closed issue #426: [Bug] CheckPoint Timeout Due to Retrieving Schemas for All Subpartitions When Synchronizing a Partitioned PostgreSQL Table
URL: https://github.com/apache/doris-flink-connector/issues/426
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [I] [Bug] CheckPoint Timeout Due to Retrieving Schemas for All Subpartitions When Synchronizing a Partitioned PostgreSQL Table [doris-flink-connector]
Posted by "DongLiang-0 (via GitHub)" <gi...@apache.org>.
DongLiang-0 commented on issue #426:
URL: https://github.com/apache/doris-flink-connector/issues/426#issuecomment-2236150326
@tchivs Thanks for you raising this issue.
This seems to be an issue with flink-cdc-connector. You can try contributing your fix to the flink-cdc community.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org