zhang-arvin commented on issue #11847: URL: https://github.com/apache/seatunnel/issues/11847#issuecomment-5371865951
@SEZ9 Thanks for the clear scope. I have completed the analysis and submitted a fix in PR #11864. Here is a summary of the findings and approach for each of your three asks: ### 1. Reproduction analysis I did not reproduce the race locally with a timing hook — instead I used static code analysis to trace the full savepoint→restore path, and found **three deterministic bugs** (none require a timing hook to trigger — they are logic errors that make the race *inevitable* rather than accidental): **Bug 1 — `commitCurrentOffset` sends the wrong LSN to Debezium** `PostgresWalFetchTask.commitCurrentOffset()` calls `offset.getLsn()` which reads from `SourceInfo.LSN_KEY` (`"lsn"`), but Debezium's `commitOffset` expects the `LAST_COMMIT_LSN_KEY` (`"lsn_commit"`) field. This means the slot's confirmed LSN is advanced to the mid-transaction LSN of the last processed event, not its commit boundary — so on a subsequent restart, the slot has already moved past rows that committed between the savepoint and the stop. **Bug 2 — `shouldEmit` uses strict `isAfter` instead of `isAtOrAfter`** `IncrementalSourceStreamFetcher.shouldEmit()` uses `isAfter(splitStartWatermark)` (strict greater-than). After recovery, the first event with the same LSN as `splitStartWatermark` is filtered out — even though that event was never delivered before the savepoint. **Bug 3 — `execute()` never checks `errorHandler.getProducerThrowable()`** `PostgresWalFetchTask.execute()` does not check `errorHandler.getProducerThrowable()` after the streaming loop returns. Debezium 1.9.8 swallows streaming failures into the error handler, so a wedged replay after restore is silently invisible to the job. These three bugs together explain the symptom perfectly: the offset commit advances the slot past the row, the emit filter drops the row on recovery, and the error handler never surfaces the silent gap. This is consistent with @davidzollo's evidence that two unrelated branches reproduce the failure — the bugs are in the shared CDC base path, not in any single PR's deltas. ### 2. Fix approach (PR #11864) I adopted a **hybrid A+B approach**: - **A (LSN key fix):** Added `getLsnCommit()` to `LsnOffset` that reads from `LAST_COMMIT_LSN_KEY` with fallback to `LSN_KEY`, and updated `commitCurrentOffset` to use the committed LSN. Also fixed the `LsnOffset` constructor to store `LAST_COMMIT_LSN_KEY` (it was missing). - **B (emit boundary fix):** Changed `shouldEmit` from `isAfter` to `isAtOrAfter` so the watermark-position event is included on recovery. - **Error handling:** Added `errorHandler.getProducerThrowable()` check in `execute()` to surface streaming failures. ### 3. Tests - **`LsnOffsetTest`**: 4 deterministic unit tests covering `getLsnCommit()` fallback logic and verifying that the constructor stores `LAST_COMMIT_LSN_KEY`. - **`IncrementalSourceStreamFetcherTest`**: 5 boundary tests covering the `isAtOrAfter`/`isAfter` comparison, including the exact watermark-position case. All tests pass. The full diff is in #11864 — happy to adjust the approach based on your feedback. -- 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]
