hudi-agent commented on code in PR #19971:
URL: https://github.com/apache/hudi/pull/19971#discussion_r4043934526
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java:
##########
@@ -151,6 +152,17 @@ public void restartHeartbeat(String instantTime) {
if (isStreamingWriteMetadataTable) {
this.streamingMetadataWriteHandler.startHeartbeat(instantTime,
getHoodieTable());
}
+ // 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();
+ long waitStarted = System.nanoTime();
+ if (getConfig().getFailedWritesCleanPolicy().isLazy()) {
+ getHeartbeatClient().awaitHeartbeat(instantTime, timeoutMs);
+ }
+ if (isStreamingWriteMetadataTable) {
+ long remainingMs = Math.max(0, timeoutMs -
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted));
+ this.streamingMetadataWriteHandler.awaitHeartbeat(instantTime,
remainingMs);
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java:
##########
@@ -151,6 +152,17 @@ public void restartHeartbeat(String instantTime) {
if (isStreamingWriteMetadataTable) {
this.streamingMetadataWriteHandler.startHeartbeat(instantTime,
getHoodieTable());
}
+ // 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();
Review 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.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java:
##########
@@ -144,14 +146,45 @@ public void start(String instantTime) {
Heartbeat newHeartbeat = new Heartbeat();
newHeartbeat.setHeartbeatStarted(true);
instantToHeartbeatMap.put(instantTime, newHeartbeat);
- // Ensure heartbeat is generated for the first time with this blocking
call.
- // 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().
Review 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.
--
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]
