Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19971 +/- ##
============================================
+ Coverage 80.22% 80.32% +0.09%
- Complexity 34669 34751 +82
============================================
Files 2546 2545 -1
Lines 142452 142522 +70
Branches 17330 17505 +175
============================================
+ Hits 114281 114474 +193
+ Misses 20265 20143 -122
+ Partials 7906 7905 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
| // Since scheduler submits the task to a thread, no guarantee when that thread will get CPU | ||
| // cycles to generate the first heartbeat. | ||
| // Attempt the first heartbeat synchronously. A timed-out write is retried by the scheduler; | ||
| // callers that need a confirmed heartbeat before proceeding can use awaitHeartbeat(). |
There was a problem hiding this comment.
didn't quite get this, the heartbeat is updated instantly in memory right? and the heartbeat validaiton always goes through heartbeatClient.isHeartbeatExpired(instantTime).
There was a problem hiding this comment.
The in-memory lastHeartbeatTime is updated only after writeHeartbeatFile() succeeds. If the initial write times out (no exception will be thrown), it remains null, while start() schedules retries and returns normally.
You’re right that validation calls isHeartbeatExpired(), but with a null timestamp it falls back to storage. If no heartbeat file exists, it reads 0 and treats the heartbeat as expired.
See comment here:
There was a problem hiding this comment.
the updateHeartbeat(String instantTime) already sets up the heartbat time through heartbeat.setLastHeartbeatTime(newHeartbeatTime);
You’re right that validation calls isHeartbeatExpired(), but with a null timestamp
this is the reader hearbeat behavior but since the commit happens on the coordinator which shares the same heartbeat client which got updated from writer heartbeat.
There was a problem hiding this comment.
the updateHeartbeat(String instantTime) already sets up the heartbat time through heartbeat.setLastHeartbeatTime(newHeartbeatTime);
The timestamp is assigned only after the storage write succeeds without timeout exception. The relevant execution order in updateHeartbeat() is:
try {
writeHeartbeatFile(instantTime); // May throw TimeoutException.
// ...
heartbeat.setLastHeartbeatTime(newHeartbeatTime); // Skipped on timeout.
} catch (TimeoutException te) {
// Log the timeout without updating the timestamp; retry on the next tick.
}If the initial write times out, updateHeartbeat() catches the exception and returns normally, leaving lastHeartbeatTime null. start() then schedules retries and returns.
The coordinator can proceed directly to recommit before a retry succeeds. Even though it uses the same heartbeat client, the timestamp has never been set. isHeartbeatExpired() therefore falls back to storage; if the heartbeat file is absent, it reads 0 and rejects the commit:
Caused by: org.apache.hudi.exception.HoodieException:
Heartbeat for instant XXX has expired, last heartbeat 0
at org.apache.hudi.client.heartbeat.WriterHeartbeatUtils.abortIfHeartbeatExpired(WriterHeartbeatUtils.java:101)
at org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:277)
at org.apache.hudi.client.BaseHoodieWriteClient.commitStats(BaseHoodieWriteClient.java:252)
There was a problem hiding this comment.
🤖 One data point that may help here: the gap is new since #18904. Before that writeHeartbeatFile() was a plain blocking storage.create, so start() returned only after the write landed (or threw) and the in-memory timestamp was always set on the coordinator's client. With the bounded future.get(heartbeatWriteTimeoutMs) + cancel(true), a timed-out first write is swallowed in the TimeoutException branch of updateHeartbeat() without setLastHeartbeatTime(), so start() can now return with the map entry present but lastHeartbeatTime == null, and isHeartbeatExpired() falls back to storage where the file may not exist yet.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The change adds HoodieHeartbeatClient.awaitHeartbeat() and has Flink recommit wait for confirmed data-table and metadata-table heartbeat writes before committing, so a timed-out first write no longer surfaces as last heartbeat 0. The core wait loop looks correct; the two inline comments are about the Flink caller — the size of the wait budget on the coordinator's main-thread entry points, and the metadata-table budget being derived from the data-table config even though the MDT heartbeat client does not inherit those settings. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
| } | ||
| if (isStreamingWriteMetadataTable) { | ||
| long remainingMs = Math.max(0, timeoutMs - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted)); | ||
| this.streamingMetadataWriteHandler.awaitHeartbeat(instantTime, remainingMs); |
There was a problem hiding this comment.
🤖 The budget passed here comes from the data-table config, but the MDT heartbeat client runs on the MDT write config, and HoodieMetadataWriteUtils.createMetadataWriteConfig doesn't copy hoodie.client.heartbeat.interval_in_ms / tolerable.misses across, so the MDT scheduler always retries on the 60s default. If a user sets a shorter data-table interval (say 10s x 10 = 100s budget, or anything under 60s), the MDT retry can't fire inside the window and this wait times out every time — the exact scenario the PR is fixing. Would it make sense to compute the MDT budget from metadataWriter.getWriteClient().getConfig() inside FlinkStreamingMetadataWriteHandler.awaitHeartbeat instead of sharing the data-table window?
| } | ||
| // Recommit has no data-writing phase in which a timed-out first heartbeat can recover. | ||
| // Start both heartbeats before waiting and share the configured heartbeat expiry window. | ||
| long timeoutMs = getConfig().getHoodieClientHeartbeatIntervalInMs() * getConfig().getHoodieClientHeartbeatTolerableMisses(); |
There was a problem hiding this comment.
🤖 With defaults this is 60s x 10 = 10 minutes, and restartHeartbeat runs on the JobManager main thread via StreamWriteOperatorCoordinator.start()/resetToCheckpoint() → restoreEvents, and via subtaskReset() → executeSync. The pre-existing synchronous first write already blocks up to 60s per heartbeat, but adding up to 10 more minutes on the coordinator's main-thread entry points seems like it could stall the JM (RPC main thread, TM heartbeat handling) well past Flink's default heartbeat.timeout. Was a smaller bound — e.g. one or two heartbeat intervals, enough for the first scheduled retry — considered? @danny0405 curious whether you're comfortable with a wait of this size on these coordinator callbacks.
| // Since scheduler submits the task to a thread, no guarantee when that thread will get CPU | ||
| // cycles to generate the first heartbeat. | ||
| // Attempt the first heartbeat synchronously. A timed-out write is retried by the scheduler; | ||
| // callers that need a confirmed heartbeat before proceeding can use awaitHeartbeat(). |
There was a problem hiding this comment.
🤖 One data point that may help here: the gap is new since #18904. Before that writeHeartbeatFile() was a plain blocking storage.create, so start() returned only after the write landed (or threw) and the in-memory timestamp was always set on the coordinator's client. With the bounded future.get(heartbeatWriteTimeoutMs) + cancel(true), a timed-out first write is swallowed in the TimeoutException branch of updateHeartbeat() without setLastHeartbeatTime(), so start() can now return with the map entry present but lastHeartbeatTime == null, and isHeartbeatExpired() falls back to storage where the file may not exist yet.
5 participants
Describe the issue this Pull Request addresses
Closes #19947.
During Flink recovery, recommit can reach heartbeat validation before a timed-out initial heartbeat write succeeds on a scheduled retry. If the file is absent, commit fails with
last heartbeat 0. This affects both data-table and streaming metadata-table heartbeats.Summary and Changelog
HoodieHeartbeatClient.awaitHeartbeat()with bounded waiting based on the existinglastHeartbeatTime, made volatile for visibility across threads.start()semantics: initial write timeouts still defer to scheduled retries. Waiting does not replace commit-time expiry checks.Impact
Flink recommit waits for confirmed heartbeat writes before attempting commit and reports an explicit timeout if they remain unavailable. Adds an optional client API without new configuration. The shared waiting budget applies after heartbeat startup.
Risk Level
Low. Recovery can wait longer during storage delays, bounded by the existing heartbeat interval and tolerable-misses settings. Regression tests exercise timeout recovery and both heartbeat paths; existing heartbeat expiry validation remains in place.
Documentation Update
Added API Javadoc documenting readiness, timeout behavior, and the requirement to retain commit-time expiry validation.
Contributor's checklist