wuhainan opened a new issue, #12212: URL: https://github.com/apache/seatunnel/issues/12212
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened A long-running MySQL CDC job failed while deserializing MySQL binlog row events. The innermost exception was an `EOFException`, indicating that the event data ended before all row fields could be read. The issue occurred multiple times with different row event types: - `EXT_DELETE_ROWS`: failed while deserializing a `DATETIME_V2` column. - `EXT_WRITE_ROWS`: failed while deserializing a `LONGLONG` column. The job was automatically restarted from the latest completed checkpoint and continued running. However, two source rows created immediately before the failed event were not present in the target table when checked, indicating possible data loss around the failed binlog event. ### SeaTunnel Version 2.3.13 ### SeaTunnel Config ```conf env = { parallelism = 1 job.mode = "STREAMING" job.name = "mysql2tidb_quote_data" checkpoint.interval = 30000 checkpoint.timeout = 600000 checkpoint.mode = "EXACTLY_ONCE" restart-strategy = "failure-rate" restart-strategy.failure-rate.max-failures-per-interval = 10 restart-strategy.failure-rate.failure-rate-interval = "300 s" restart-strategy.failure-rate.delay = "10 s" jobmanager.memory.process.size = "1024m" taskmanager.memory.process.size = "2048m" taskmanager.numberOfTaskSlots = 1 } source = { MySQL-CDC = { hostname = "******" port = 3307 url = "jdbc:mysql://*****:3307/" username = "*****" password = "******" table-names-config = [ { table = "quote_data.t_contract" primaryKeys = [ "id" ] }, { table = "quote_data.t_contract_snapshot" primaryKeys = [ "id" ] }, { table = "quote_data.t_financial" primaryKeys = [ "id" ] }, { table = "quote_data.t_package" primaryKeys = [ "id" ] }, { table = "quote_data.t_package_snapshot" primaryKeys = [ "id" ] } ] database-names = [ "quote_data" ] table-names = [ "quote_data.t_contract", "quote_data.t_contract_snapshot", "quote_data.t_financial", "quote_data.t_package", "quote_data.t_package_snapshot" ] server-id = 5720 startup.mode = "initial" server-time-zone = "Asia/Shanghai" schema-changes.enabled = true plugin_output = "src_tidb_0" } } sink = { Jdbc = { plugin_input = [ "src_tidb_0" ] driver = "com.mysql.cj.jdbc.Driver" url = "jdbc:mysql://******:4001/localtest?useSSL=false&allowPublicKeyRetrieval=true&useUnicode=true&characterEncoding=utf8" username = "*****" password = "******" batch_size = 1000 database = "localtest" table = "${table_name}" generate_sink_sql = true primary_keys = [ "${primary_key}" ] enable_upsert = true schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" data_save_mode = "APPEND_DATA" multi_table_sink_replica = 1 } } ``` ### Running Command ```shell # The job is submitted by an internal platform using Flink YARN Application mode. # The equivalent command is: $FLINK_HOME/bin/flink run-application \ -t yarn-application \ -c org.apache.seatunnel.core.starter.flink.SeaTunnelFlink \ /path/to/seatunnel-flink-20-starter.jar \ --config /path/to/v2.conf ``` ### Error Exception ```log java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:147) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:167) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:93) at org.apache.seatunnel.connectors.cdc.base.source.reader.IncrementalSourceReader.pollNext(IncrementalSourceReader.java:119) Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:165) at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:81) Caused by: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. Caused by: io.debezium.DebeziumException: Failed to deserialize data of EventHeaderV4{ timestamp=1788861881000, eventType=EXT_WRITE_ROWS, serverId=22732, headerLength=19, dataLength=8170, nextPosition=350563202, flags=0 } Caused by: com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException: Failed to deserialize data of EventHeaderV4{ timestamp=1788861881000, eventType=EXT_WRITE_ROWS, serverId=22732, headerLength=19, dataLength=8170, nextPosition=350563202, flags=0 } Caused by: java.io.EOFException: Failed to read next byte from position -649582912 at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:213) at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readLong(ByteArrayInputStream.java:66) at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeLongLong(AbstractRowsEventDataDeserializer.java:242) at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeCell(AbstractRowsEventDataDeserializer.java:163) at com.github.shyiko.mysql.binlog.event.deserialization.AbstractRowsEventDataDeserializer.deserializeRow(AbstractRowsEventDataDeserializer.java:143) at com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer.deserializeRows(WriteRowsEventDataDeserializer.java:64) at com.github.shyiko.mysql.binlog.event.deserialization.WriteRowsEventDataDeserializer.deserialize(WriteRowsEventDataDeserializer.java:56) at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:335) ``` ### Zeta or Flink or Spark Version flink 1.20 ### Java or Scala Version java 17 ### Screenshots _No response_ ### Are you willing to submit PR? - [ ] 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
