danny0405 opened a new pull request, #19952: URL: https://github.com/apache/hudi/pull/19952
### Describe the issue this Pull Request addresses Issue: #19922. After restoring a Flink job under a new JobID or on attempt 0, writers currently keep checkpoint ID `-1`. If scale-up leaves a restored pending instant buffered under `-1`, new writes reuse that instant instead of creating a new one. The coordinator can then reject the write event with `Receive an unexpected event for instant` because its current instant was never initialized. Use the restored checkpoint ID for new writes so they do not collide with the restored batch. This addresses the instant-ID collision; the separate recovery gap when another restart interrupts deferred recommit remains open. ### Summary and Changelog - Initialize the writer checkpoint ID from restored state regardless of JobID or attempt number; remove the auxiliary `job-id-state` and its initialization checks. - Add writer tests for fresh starts, retries, changed or missing JobID state, checkpoint advancement, and replay of metadata with its original checkpoint ID. - Extend coordinator tests to verify actual recommit during startup and global failover, including legacy checkpoints, and verify deferred commits after scale-up from 2 to 4 writers. - Simplify coordinator setup and check restored buffers by checkpoint ID. Retain a disabled regression for metadata loss after a second restart before deferred commit. No code was copied from external sources. ### Impact Restored Flink writers request new instants using the restored checkpoint ID. Fresh starts continue to use `-1`. No public API, configuration, or Hudi table-format changes; writer snapshots no longer persist the auxiliary JobID state. ### Risk Level medium This changes writer recovery behavior. Validation on Flink 2.2.1 with Java 17: - `TestAbstractStreamWriteFunction`: 9 passed. - `TestStreamWriteOperatorCoordinator`: 41 passed, 1 intentionally disabled. - Maven compilation, Checkstyle, license checks, and `git diff --check` passed. The disabled `testDeferredRecommitSurvivesAnotherRestart` was run explicitly and reproduced the known failure: after the next checkpoint succeeds but before its deferred commit, another restart loses the original batch's write metadata. This PR does not fix that gap or claim complete recovery coverage across all Flink versions. ### Documentation Update none — this corrects internal checkpoint recovery without introducing a feature, configuration, or API change. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable -- 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]
