[Bug2.3.4] [postgresqlCDC]对源表做update操作时 ,同步任务报错停止When updating the source table, the synchronization task stops with an error #6407
Comments
2606090723
changed the title
|
This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs. |
|
+1 |
|
对源表delete时也有该问题，pgsql版本15, driver版本42.7.3 |
|
Please check your database settings Here are the steps to enable CDC (Change Data Capture) in PostgreSQL: Ensure the wal_level is set to logical: Modify the postgresql.conf configuration file by adding "wal_level = logical", restart the PostgreSQL server for the changes to take effect. Alternatively, you can use SQL commands to modify the configuration directly: ALTER SYSTEM SET wal_level TO 'logical';
SELECT pg_reload_conf();
-- Change the REPLICA policy of the specified table to FULL
ALTER TABLE your_table_name REPLICA IDENTITY FULL; |
I guess the settings are correct. |
|
check |
Search before asking
What happened
Exception in thread "main" org.apache.seatunnel.core.starter.exception.CommandExecuteException: SeaTunnel job executed failed
at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:199)
at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40)
at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:34)
Caused by: org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException: java.lang.NullPointerException
at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializationConverters.convert(SeaTunnelRowDebeziumDeserializationConverters.java:88)
at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.extractBeforeRow(SeaTunnelRowDebeziumDeserializeSchema.java:222)
at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.deserializeDataChangeRecord(SeaTunnelRowDebeziumDeserializeSchema.java:189)
at org.apache.seatunnel.connectors.cdc.debezium.row.SeaTunnelRowDebeziumDeserializeSchema.deserialize(SeaTunnelRowDebeziumDeserializeSchema.java:111)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceRecordEmitter.emitElement(IncrementalSourceRecordEmitter.java:157)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceRecordEmitter.processElement(IncrementalSourceRecordEmitter.java:132)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceRecordEmitter.emitRecord(IncrementalSourceRecordEmitter.java:89)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceRecordEmitter.emitRecord(IncrementalSourceRecordEmitter.java:55)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:108)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceReader.pollNext(IncrementalSourceReader.java:110)
at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:112)
at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceReader.pollNext(IncrementalSourceReader.java:110)
at org.apache.seatunnel.engine.server.task.flow.SourceFlowLifeCycle.collect(SourceFlowLifeCycle.java:150)
at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.collect(SourceSeaTunnelTask.java:116)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.stateProcess(SeaTunnelTask.java:168)
at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.call(SourceSeaTunnelTask.java:121)
at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:643)
at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:944)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
2024-02-28 08:45:23,260 INFO [.e.FieldNamedPreparedStatement] [BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - PrepareStatement sql is:
INSERT INTO "test_db"."public"."t_user_2" ("id", "name", "age", "update_time") VALUES (?, ?, ?, ?) ON CONFLICT ("id") DO UPDATE SET "id"=EXCLUDED."id", "name"=EXCLUDED."name", "age"=EXCLUDED."age", "update_time"=EXCLUDED."update_time"
2024-02-28 08:45:23,265 INFO [.e.FieldNamedPreparedStatement] [BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - PrepareStatement sql is:
DELETE FROM "test_db"."public"."t_user_2" WHERE "id" = ?
2024-02-28 08:45:23,267 INFO [.JdbcMultiTableResourceManager] [BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - start close connection poolHikariPool-4
2024-02-28 08:45:23,268 INFO [o.a.s.s.c.z.h.HikariDataSource] [BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - HikariPool-4 - Shutdown initiated...
2024-02-28 08:45:23,272 WARN [o.a.s.s.c.z.h.HikariDataSource] [BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - HikariPool-4 - Interrupted during closing
java.lang.InterruptedException: null
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2067) ~[?:1.8.0_342]
at java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1475) ~[?:1.8.0_342]
at org.apache.seatunnel.shade.com.zaxxer.hikari.pool.HikariPool.shutdown(HikariPool.java:255) ~[?:?]
at org.apache.seatunnel.shade.com.zaxxer.hikari.HikariDataSource.close(HikariDataSource.java:351) ~[?:?]
at org.apache.seatunnel.connectors.seatunnel.jdbc.sink.ConnectionPoolManager.close(ConnectionPoolManager.java:66) ~[?:?]
at org.apache.seatunnel.connectors.seatunnel.jdbc.sink.JdbcMultiTableResourceManager.close(JdbcMultiTableResourceManager.java:42) ~[?:?]
at org.apache.seatunnel.connectors.seatunnel.common.multitablesink.MultiTableSinkWriter.close(MultiTableSinkWriter.java:251) ~[seatunnel-transforms-v2.jar:2.3.4]
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.close(SinkFlowLifeCycle.java:153) ~[seatunnel-starter.jar:2.3.4]
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$close$5(SeaTunnelTask.java:330) ~[seatunnel-starter.jar:2.3.4]
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) [?:1.8.0_342]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384) [?:1.8.0_342]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) [?:1.8.0_342]
at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:290) [?:1.8.0_342]
at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731) [?:1.8.0_342]
at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) [?:1.8.0_342]
at java.util.concurrent.ForkJoinPool.helpComplete(ForkJoinPool.java:1870) [?:1.8.0_342]
at java.util.concurrent.ForkJoinPool.externalHelpComplete(ForkJoinPool.java:2467) [?:1.8.0_342]
at java.util.concurrent.ForkJoinTask.externalAwaitDone(ForkJoinTask.java:324) [?:1.8.0_342]
at java.util.concurrent.ForkJoinTask.doInvoke(ForkJoinTask.java:405) [?:1.8.0_342]
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:734) [?:1.8.0_342]
at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:159) [?:1.8.0_342]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:173) [?:1.8.0_342]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) [?:1.8.0_342]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485) [?:1.8.0_342]
at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:650) [?:1.8.0_342]
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.close(SeaTunnelTask.java:327) [seatunnel-starter.jar:2.3.4]
at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:660) [seatunnel-starter.jar:2.3.4]
at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:944) [seatunnel-starter.jar:2.3.4]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_342]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_342]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_342]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_342]
at java.lang.Thread.run(Thread.java:750) [?:1.8.0_342]
2024-02-28 08:45:23,899 INFO [r.IncrementalSourceSplitReader] [Source Data Fetcher for BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - Close current fetcher org.apache.seatunnel.connectors.cdc.base.source.reader.external.IncrementalSourceStreamFetcher
2024-02-28 08:45:23,900 INFO [i.d.j.JdbcConnection ] [pool-64-thread-1] - Connection gracefully closed
2024-02-28 08:45:23,902 INFO [i.d.j.JdbcConnection ] [pool-65-thread-1] - Connection gracefully closed
2024-02-28 08:45:23,903 INFO [i.d.j.JdbcConnection ] [pool-66-thread-1] - Connection gracefully closed
2024-02-28 08:45:24,304 INFO [o.a.s.c.s.c.s.r.f.SplitFetcher] [Source Data Fetcher for BlockingWorker-TaskGroupLocation{jobId=815141672055734273, pipelineId=1, taskGroupId=30000}] - Split fetcher 0 exited.
2024-02-28 08:45:24,316 INFO [s.c.s.s.c.ClientExecuteCommand] [Thread-105] - run shutdown hook because get close signal
SeaTunnel Version
SeaTunnel2.3.4 pgsqlCDC
SeaTunnel Config
Running Command
root@7366e5930d9f:/opt/apache-seatunnel-2.3.4# ./bin/seatunnel.sh --config ./config/pgsqlcdc.template -e localError Exception
Zeta or Flink or Spark Version
zeta
Java or Scala Version
No response
Screenshots
No response
Are you willing to submit PR?
Code of Conduct
The text was updated successfully, but these errors were encountered: