[I] Flink recommit can fail when the initial heartbeat write times out [hudi]

["cshuo (via GitHub)" <gi...@apache.org>]

cshuo opened a new issue, #19947:
URL: https://github.com/apache/hudi/issues/19947

   ### Description
   
   During Flink recovery, recommitting a pending instant can fail if the initial heartbeat write times out and commit proceeds before a background retry succeeds.
   
   This can affect both **data-table and metadata-table heartbeats** when the table uses the `LAZY` failed-writes cleaning policy. The reported failure occurred during a metadata-table commit, even with #19237 applied.
   
   `HoodieHeartbeatClient.start()` attempts the first heartbeat write synchronously. On timeout, `updateHeartbeat()` logs the failure without recording a successful heartbeat. `start()` then schedules background retries and returns normally.
   
   This behavior intentionally allows recovery from transient storage delays. However, Flink recommit proceeds directly to commit using restored write statuses, leaving little time for a background retry to succeed.
   
   ### Failure sequence
   
   1. Flink restores write metadata for a pending instant.
   2. `recommitInstant()` calls `restartHeartbeat()`.
   3. The initial data-table or metadata-table heartbeat write times out.
   4. `start()` returns normally, with background retries scheduled.
   5. Recommit reaches heartbeat validation before a retry succeeds.
   6. If no heartbeat file is available, the check reads `0` and rejects the commit as expired.
   
   ### Observed error
   
   ```text
   HoodieException: Failed to start operator coordinator.
   Caused by: HoodieException: Error while completing streaming commit
   to metadata with instant 20260831111848938
   Caused by: HoodieException: Heartbeat for instant 20260831111848938
   has expired, last heartbeat 0
   ```
   
   Relevant call path:
   
   ```text
   StreamWriteOperatorCoordinator.start
     → restoreEvents
     → recommitInstant
     → commitInstant
     → commitToMetadataTable
     → completeStreamingCommit
     → commitStats
     → abortIfHeartbeatExpired
   ```
   
   The initial heartbeat write timeout was confirmed in the reported case. The reason the previous heartbeat file was absent has not been established.
   
   ### Expected behavior
   
   Before recommitting, Flink should allow a bounded opportunity for the required heartbeats to become available. If they remain unavailable, recovery should fail with an explicit heartbeat-readiness timeout.
   
   ### Suggested approach
   
   Add an optional `awaitHeartbeat(instant, timeout)` API and invoke it before Flink recommit:
   
   - Start the required data-table and metadata-table heartbeats.
   - Wait for confirmed successful heartbeat writes within a bounded budget.
   - Proceed with recommit only after the required heartbeats are ready.
   
   Preserve the existing `start()` behavior introduced by #18904, including background recovery from an initial write timeout. Retain commit-time heartbeat validation and existing timeline/rollback checks; heartbeat readiness alone does not establish that an instant is safe to recommit.
   
   ### Validation
   
   - Initial heartbeat write times out, a background retry succeeds, and recommit proceeds.
   - Persistent write failures cause a bounded wait timeout.
   - Waiting handles interruption and stopped or expired heartbeats.
   - Ordinary `start()` callers retain background retry behavior.
   - Cover both data-table and metadata-table heartbeats, including an EAGER data table with a LAZY metadata table.
   


-- 
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: commits-unsubscribe@hudi.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org