DanielLeens commented on issue #13106: URL: https://github.com/apache/seatunnel/issues/13106#issuecomment-6595075893
Thanks for isolating the Flink side. Current `dev` confirms the routing gap: nothing under `seatunnel-translation-flink` calls `SinkAggregatedCommitter#restoreCommit`. - Flink 1.13/1.15 translation (shared `seatunnel-translation-flink-common` `FlinkGlobalCommitter`): `filterRecoveredCommittables` returns an empty list (L65-69), and `commit` calls `aggregatedCommitter.commit` and discards any returned re-commit list (L77-85). - Flink 1.20 (`FlinkSimpleAggregatedCommitter#commit`): every `CommitRequest`, including ones Flink re-delivers from restored committer state, goes through `aggregatedCommitter.commit` (L138-139). Requests are signalled `signalAlreadyCommitted()` even when the committer returns items to retry (L141-156). #12807 made `MultiTableSinkAggregatedCommitter` forward `restoreCommit` per table, but only Zeta calls that entry point, so it does not cover Flink. With the strict JDBC XA commit from #11569, a Flink restore after a checkpoint whose XIDs were already committed should hit the same `XAER_NOTA` failure as #12805. That conclusion comes from reading the code; I have not run it on Flink here. The second point, acknowledging returned retry items as committed, is a separate silent-loss path for any sink that returns them, and it belongs in the same fix. #13110 is the fix PR, so further work should go there. Points I would expect review to check: 1. Savepoint compatibility: the new committable serializers (`CheckpointCommitWrapperSerializer`, `RecoveryCommitSerializer`) must read committables written by the current `CommitWrapperSerializer` of the latest release. Please add a test that uses bytes produced by the released serializer. 2. Topology: adding `RestoredCommittableOperator` must keep the existing operator uids, so a release savepoint restores without `--allowNonRestoredState`. 3. For 1.13/1.15, state what Flink does with the restored global committables today, given the empty `filterRecoveredCommittables`: are they dropped or re-committed? That decides whether the current bug is a commit failure or lost commits. 4. Recovery provenance must survive `retryLater` and partial progress, while fresh commits stay strict. 5. A native savepoint-restore E2E with JDBC XA where the restored checkpoint was already committed before the failure, run on at least one old-API version (1.13/1.15) and on 1.20. Collector restore stays in #12983, and the schema protocol stays in #11958. -- 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]
