fix: prevent heartbeat timer from being permanently killed by slow or delayed heartbeats#18904
fix: prevent heartbeat timer from being permanently killed by slow or delayed heartbeats#18904prashantwason wants to merge 3 commits into
Conversation
… delayed heartbeats HoodieHeartbeatClient could permanently stop generating heartbeats for an instant, causing later commits to abort with "Heartbeat for instant ... has expired" even though the writer was still alive: - The heartbeat file is written synchronously on the Timer thread. Because the timer uses scheduleAtFixedRate, a slow or hung storage write blocks the thread and freezes all subsequent heartbeats for that instant. - When a heartbeat refresh is delayed past the tolerable interval, updateHeartbeat() called Thread.currentThread().interrupt(), permanently killing the timer thread and turning a transient delay into a permanent blackout. Fix: - Perform the heartbeat file write on a bounded daemon executor (Future.get with a per-interval timeout) so a slow or hung storage call cannot block the timer thread; a timed-out write is retried on the next tick. - Remove the self-interrupt; log a warning and continue. The commit-time check HeartbeatUtils.abortIfHeartbeatExpired() remains the sole enforcement point. Add TestHoodieHeartbeatClient.testTimerSurvivesHungHeartbeatWrite.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for working on this! The PR addresses two real issues with the heartbeat client: (a) synchronous storage writes blocking the timer thread, and (b) the self-interrupt permanently killing the timer. The new design (executor with bounded timeout + remove self-interrupt) is reasonable. One edge case worth double-checking in the inline comments around what happens when the very first heartbeat write times out during start(). Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. One minor type inconsistency worth noting in the new field declaration.
| heartbeat.setInstantTime(instantTime); | ||
| heartbeat.setLastHeartbeatTime(newHeartbeatTime); | ||
| heartbeat.setNumHeartbeats(heartbeat.getNumHeartbeats() + 1); | ||
| } catch (TimeoutException te) { |
There was a problem hiding this comment.
🤖 If the very first writeHeartbeatFile() call (invoked synchronously from start()) hits this timeout path, heartbeat.setLastHeartbeatTime(...) is never reached, so the Heartbeat's lastHeartbeatTime stays null. A subsequent call to isHeartbeatExpired() (e.g., from HeartbeatUtils.abortIfHeartbeatExpired() during a fast write-then-commit) would NPE at line 189 (currentTime - getLastHeartbeatTime() unboxes a null Long). Could lastHeartbeatTime be initialized conservatively here (or in start() before scheduling), or isHeartbeatExpired guarded against the null case, to close this corner?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
There was a problem hiding this comment.
Good catch — fixed. isHeartbeatExpired() now falls back to the DFS read when the in-memory last-heartbeat time is null (which can happen if the first synchronous write times out), so it no longer NPEs on the unboxing comparison. A missing heartbeat file reads as 0 and is correctly treated as expired.
| // Maximum time the timer thread will wait for a single heartbeat file write to complete before | ||
| // abandoning it and letting the next tick retry. Bounded to one interval so that a slow/hung | ||
| // storage write cannot block the timer thread (and thus freeze all subsequent heartbeats). | ||
| private final long heartbeatWriteTimeoutMs; |
There was a problem hiding this comment.
🤖 nit: the two sibling duration fields (heartbeatIntervalInMs, maxAllowableHeartbeatIntervalInMs) are boxed Long — could you align heartbeatWriteTimeoutMs to Long as well to keep the field group consistent?
- AI-generated; verify before applying. React 👍/👎 to flag quality.
There was a problem hiding this comment.
Done — changed heartbeatWriteTimeoutMs to boxed Long to match the sibling duration fields.
| public void close() { | ||
| this.stopHeartbeatTimers(); | ||
| this.instantToHeartbeatMap.clear(); | ||
| synchronized (this) { |
There was a problem hiding this comment.
do we have concurernt close on this client, should we add a isClosed flag to avoid repetitive close.
There was a problem hiding this comment.
Added a closed flag — close() is now synchronized and idempotent (a second call is a no-op), and the executor getter throws if used after close so a late timer tick cannot resurrect the executor.
There was a problem hiding this comment.
Correction to my note above: CI caught that the heartbeat client is legitimately reused after close() (startCommit / acquireRollbackHeartbeatIfMultiWriter call start() again on a closed instance), so a sticky closed flag broke that and failed TestJavaHoodieBackedMetadata / TestHoodieJavaClientOnCopyOnWriteStorage. I removed the flag. close() is now idempotent via synchronized + the executor null-check (repeated/concurrent close is a no-op), while reuse-after-close still works (the executor is lazily re-created on the next heartbeat). Fixed in 165a2fc.
| // future heartbeats for this instant, turning a transient delay into a permanent blackout. | ||
| // Enforcement is done at commit time in HeartbeatUtils.abortIfHeartbeatExpired(), which is the | ||
| // correct and sole enforcement point. | ||
| log.warn("Missed generating heartbeat for instant {} within allowable interval {} ms; continuing to refresh", |
There was a problem hiding this comment.
makes sense somehow, but I do see some risk for correctness: when failed writes rollback strategy is configured as LAZY, the async cleaner would possibility rollback the current instant by removing some data files(not remove the metadata files on timeline yet), and then the write finish to commit, then the commit got data loss.
Should we also increase the tolerale missing cnt, the current default is 2, should we change it to 10 or 20
There was a problem hiding this comment.
Thanks, this is an important catch. I reworked the expiry handling to preserve the original safety property:
- On a detected lapse we now stop refreshing the heartbeat (cancel the timer) and do not advance the last-heartbeat time. So a lapsed writer still fails at commit via
abortIfHeartbeatExpired()— it cannot resurrect a heartbeat that an async cleaner (LAZY policy) may already have acted on, which closes the data-loss window you described. The only change from the original is that the timer is cancelled cleanly instead of viaThread.currentThread().interrupt()(which permanently killed the timer thread on the first miss). - Separately, I raised the default
hoodie.client.heartbeat.tolerable.missesfrom 2 to 10 as you suggested, so transient GC/storage pauses are far less likely to trip expiry in the first place. Combined with moving the heartbeat write off the timer thread (time-bounded), transient delays no longer cause spurious aborts, while a genuine lapse still aborts safely.
- isHeartbeatExpired: handle a null last-heartbeat-time (the very first write can time out, leaving it unset) by falling back to the DFS read, avoiding an NPE on the unboxing comparison. - On a detected lapse, stop refreshing the heartbeat (cancel the timer) and do not advance the last-heartbeat time, so a lapsed writer still aborts at commit via HeartbeatUtils.abortIfHeartbeatExpired() and cannot resurrect a heartbeat that a concurrent cleaner (LAZY failed-writes policy) may already have acted on. The timer is cancelled cleanly rather than via Thread.interrupt(). - close(): make idempotent via a closed flag and guard executor creation after close. - Use boxed Long for heartbeatWriteTimeoutMs to match the sibling duration fields. - Raise the default hoodie.client.heartbeat.tolerable.misses from 2 to 10 so transient driver pauses (e.g. GC) or storage-latency spikes do not abort a still-healthy writer.
A write client reuses its HoodieHeartbeatClient across operations: after close(), startCommit() and acquireRollbackHeartbeatIfMultiWriter() call heartbeatClient.start() again. The previously added "already closed" guard in the executor accessor turned this valid reuse into a failure (observed in TestJavaHoodieBackedMetadata and TestHoodieJavaClientOnCopyOnWriteStorage). Remove the guard and the closed flag; close() remains idempotent via synchronization plus the null-check on the executor, so repeated/concurrent close() is still safe while reuse-after-close works (the executor is lazily re-created on the next heartbeat).
| // kill the timer thread (turning a transient delay into a permanent blackout on the first miss). | ||
| log.error("Missed generating heartbeat for instant {} within allowable interval {} ms; stopping heartbeat refresh", | ||
| instantTime, this.maxAllowableHeartbeatIntervalInMs); | ||
| heartbeat.getTimer().cancel(); |
There was a problem hiding this comment.
can we replace the Timer with a Java ScheduledExecutorService, java.util.Timer. Timer is a legacy class (introduced in Java 1.3) with severe design flaws. If a TimerTask throws an uncaught exception, the entire timer thread dies, killing all other scheduled tasks permanently.
|
CI on
None of these touch Could a committer please re-run the two failed |
|
@prashantwason do you think this is a valuable suggestion: #18904 (comment) |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #18904 +/- ##
============================================
- Coverage 68.81% 68.79% -0.02%
+ Complexity 29160 29097 -63
============================================
Files 2520 2517 -3
Lines 140056 139848 -208
Branches 17209 17213 +4
============================================
- Hits 96373 96206 -167
+ Misses 35909 35867 -42
- Partials 7774 7775 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
🤖 This review was generated by an AI agent and may contain mistakes. Please verify any suggestions before applying.
Thanks for the iteration — the rollback-safety reasoning you added to the in-flight expiry path and the DFS fallback in isHeartbeatExpired() both look well thought through. One question about whether the cleaner-rollback safety property is fully preserved when the very first writeHeartbeatFile from start() times out and recovery only happens after the tolerable window. Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here.
| outputStream.close(); | ||
| writeHeartbeatFile(instantTime); | ||
| Heartbeat heartbeat = instantToHeartbeatMap.get(instantTime); | ||
| if (heartbeat.getLastHeartbeatTime() != null && isHeartbeatExpired(instantTime)) { |
There was a problem hiding this comment.
🤖 The comment below correctly explains why we must stop refreshing once a lapse has been observed — but this lastHeartbeatTime != null guard means the same protection isn't applied to the first-write-timeout case. If the synchronous writeHeartbeatFile from start() times out and the subsequent ticks also time out for longer than maxAllowableHeartbeatIntervalInMs before one finally lands, a concurrent LAZY-policy cleaner could have rolled back this instant in the gap (no file on DFS → static HoodieHeartbeatUtils.isHeartbeatExpired returns expired → getInstantsToRollbackForLazyCleanPolicy picks it up). When a tick finally succeeds, this guard skips the expiry check and sets a fresh lastHeartbeatTime, so the commit-time abortIfHeartbeatExpired() then sees a healthy in-memory time and proceeds — exactly the "commit on top of rolled-back files" outcome the comment is trying to prevent. Was this edge case considered? One option is to also abort if lastHeartbeatTime is null and currentTime - <start()-time> already exceeds maxAllowableHeartbeatIntervalInMs.
- AI-generated; verify before applying. React 👍/👎 to flag quality.
5 participants
Describe the issue this Pull Request addresses
Closes #18903
HoodieHeartbeatClientcan permanently stop generating heartbeats for an in-flight instant, causing a later commit to abort withHoodieException: Heartbeat for instant <t> has expiredeven though the writer is still alive. Two independent causes, both inupdateHeartbeat():Timerthread. Since the timer usesscheduleAtFixedRate, a slow or hung storage write blocks the thread and freezes all subsequent heartbeats for the instant.updateHeartbeat()callsThread.currentThread().interrupt(), which permanently kills the timer thread — turning a transient delay (GC pause, driver stall, single slow write) into a permanent blackout.Summary and Changelog
Future.get(heartbeatWriteTimeoutMs)), so a slow or hung storage call can no longer block the timer thread. The write timeout is one heartbeat interval; a timed-out write does not advance the last-heartbeat time and is retried on the next tick. A cached thread pool is used so that if one write hangs, subsequent ticks proceed on a fresh thread.updateHeartbeat(). Instead ofThread.currentThread().interrupt(), log a warning and continue refreshing. The commit-time checkHeartbeatUtils.abortIfHeartbeatExpired()remains the sole enforcement point for staleness.close().TestHoodieHeartbeatClient.testTimerSurvivesHungHeartbeatWrite, which blocks the first heartbeat write and asserts the timer keeps generating heartbeats (covering both fixes).Impact
No public API or config change. Heartbeat refresh becomes resilient to transient storage latency and driver pauses: a transient stall no longer permanently disables heartbeats for an instant. Staleness is still enforced at commit time, so correctness of the concurrency guard is unchanged.
Risk Level
low
Behavior change is confined to
HoodieHeartbeatClient. ExistingTestHoodieHeartbeatClienttests pass and a new regression test was added.Documentation Update
none
Contributor's checklist