xinnyuli opened a new issue, #12624: URL: https://github.com/apache/seatunnel/issues/12624
### 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 When checkpointing is on, the JDBC sink commits only at checkpoint time (prepareCommit), on a timer flush, or on close. If the connection uses manual commit (auto_commit = false, or Oracle, which the sink always switches to manual commit), every batch-size flush between two commits goes into the same open transaction. If that transaction is lost before the commit, the earlier flushed batches are gone. The sink does not notice, and the next checkpoint still succeeds. I found two ways this happens. 1. Flush retry (needs max_retries > 0) Batch A is flushed into the open transaction, but not committed yet. The flush of batch B fails. Either the connection drops (SQLState 08xxx), or the database rolls back the whole transaction (SQLState 40001, for example a deadlock). JdbcOutputFormat.flush reconnects if needed and retries. The retry only re-sends batch B, because batch A was already cleared from the executor buffer after its own successful flush. The checkpoint commits. Only batch B is in the database. Batch A is lost, and there is no error. 2. Pooled connection replaced before commit (default config) ConnectionPoolManager.getConnection(index) silently replaces a cached connection that is no longer usable. In Zeta every JDBC sink runs through MultiTableSink, even with a single table, so this pool is the normal path. Batches are flushed into the open transaction on connection C1. C1 dies before the checkpoint, for example from a network blip, a DB failover, or an idle timeout. The database rolls back its transaction. In prepareCommit, commitIfNeeded() calls connectionProvider.getConnection(), gets a new connection C2, and commits it. C2 has nothing to commit, so the commit succeeds and the checkpoint succeeds. Everything flushed on C1 since the last checkpoint is lost, and there is no error. In both cases the rows are not replayed. The checkpoint succeeded, so the source offsets moved past them. With auto_commit = true (the default for most databases) each batch is committed by the driver right away, so neither path loses data. ### SeaTunnel Version dev (checked at af80a6704be4734b20a913f46f3838c3770688eb) ### SeaTunnel Config ```conf env { parallelism = 1 job.mode = "STREAMING" checkpoint.interval = 60000 } source { # any source } sink { Jdbc { url = "jdbc:postgresql://localhost:5432/test" driver = "org.postgresql.Driver" user = "postgres" password = "postgres" database = "test" table = "public.t" generate_sink_sql = true primary_keys = ["id"] auto_commit = false batch_size = 1000 max_retries = 3 # path 1 only; path 2 also happens with the default 0 } } ``` ### Running Command ```shell ./bin/seatunnel.sh --config jdbc_sink.conf -m local To trigger it, kill the sink's database session after the first batch flush and before the next checkpoint. On Postgres you can find it in pg_stat_activity (state idle in transaction) and end it with pg_terminate_backend(pid). ``` ### Error Exception ```log None. That is the problem: the job keeps running and the checkpoint succeeds. With max_retries > 0 the only trace is one "JDBC executeBatch error, retry times = 0" ERROR line, followed by a successful retry. ``` ### Zeta or Flink or Spark Version Zeta (dev) ### Java or Scala Version 1.8 ### Screenshots Deterministic reproduction with SQLite (no Docker needed). Both tests fail on dev, because the checkpoint succeeds and rows are lost: JdbcMultiTableReconnectTest#manualCommitWriterDoesNotDropEarlierBatchWhenLaterBatchReconnects: path 1. JdbcMultiTableReconnectTest#manualCommitWriterFailsCheckpointWhenConnectionIsReplacedBeforeCommit: path 2. The test provider swaps a dead connection on getConnection(), the same way ConnectionPoolManager does. ### 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
