hudi-agent commented on code in PR #19867:
URL: https://github.com/apache/hudi/pull/19867#discussion_r3958654759
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/WriterHeartbeatUtils.java:
##########
@@ -58,8 +59,10 @@ public static boolean deleteHeartbeatFile(HoodieStorage
storage,
} else {
log.info("Deleted the heartbeat for instant {}", instantTime);
}
- } catch (IOException io) {
- log.error("Unable to delete heartbeat for instant {}", instantTime, io);
+ } catch (IOException | HoodieIOException e) {
Review Comment:
🤖 nit: since `HoodieIOException` is unchecked, catching it alongside the
checked `IOException` here reads a bit oddly — might be worth a short comment
(or a second catch block) explaining it's deliberately covering both the
wrapped IOException path and the unchecked storage rejection.
<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:
##########
@@ -196,15 +196,34 @@ private boolean isHeartbeatStarted(Heartbeat heartbeat) {
private void stopHeartbeatScheduler(Heartbeat heartbeat) {
log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime());
shutdownHeartbeatScheduler(heartbeat);
+ // Callers delete the heartbeat file next. A refresh landing after that
delete recreates the
+ // file, and on storage that enforces preconditions it also changes the
object generation, so a
+ // generation-matched delete is rejected (e.g. GCS 412 conditionNotMet).
+ awaitHeartbeatSchedulerTermination(heartbeat);
heartbeat.setHeartbeatStopped(true);
log.info("Stopped heartbeat for instant {}", heartbeat.getInstantTime());
}
+ /** Stops further refreshes without waiting; safe to call from the scheduler
thread itself. */
private void shutdownHeartbeatScheduler(Heartbeat heartbeat) {
if (heartbeat.getScheduledFuture() != null) {
heartbeat.getScheduledFuture().cancel(false);
}
- heartbeat.getHeartbeatScheduler().shutdownNow();
+ heartbeat.getHeartbeatScheduler().shutdown();
Review Comment:
🤖 Since the tick is now allowed to run to completion, have you considered
that `stop()` has already done `instantToHeartbeatMap.remove(instantTime)`
before it awaits? The in-flight `updateHeartbeat` then hits
`instantToHeartbeatMap.get(instantTime)` → null → NPE at
`heartbeat.getLastHeartbeatTime()`, which `HeartbeatTask` logs at ERROR as
"will retry on next tick" on the success path of a durable commit. A `heartbeat
== null` early-return in `updateHeartbeat` would keep that quiet.
<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:
##########
@@ -196,15 +196,34 @@ private boolean isHeartbeatStarted(Heartbeat heartbeat) {
private void stopHeartbeatScheduler(Heartbeat heartbeat) {
log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime());
shutdownHeartbeatScheduler(heartbeat);
+ // Callers delete the heartbeat file next. A refresh landing after that
delete recreates the
+ // file, and on storage that enforces preconditions it also changes the
object generation, so a
+ // generation-matched delete is rejected (e.g. GCS 412 conditionNotMet).
+ awaitHeartbeatSchedulerTermination(heartbeat);
heartbeat.setHeartbeatStopped(true);
log.info("Stopped heartbeat for instant {}", heartbeat.getInstantTime());
}
+ /** Stops further refreshes without waiting; safe to call from the scheduler
thread itself. */
private void shutdownHeartbeatScheduler(Heartbeat heartbeat) {
if (heartbeat.getScheduledFuture() != null) {
heartbeat.getScheduledFuture().cancel(false);
}
- heartbeat.getHeartbeatScheduler().shutdownNow();
+ heartbeat.getHeartbeatScheduler().shutdown();
+ }
+
+ private void awaitHeartbeatSchedulerTermination(Heartbeat heartbeat) {
+ // An in-flight tick can be parked on the bounded write, so allow for that
plus one interval.
+ long timeoutMs = this.heartbeatWriteTimeoutMs + this.heartbeatIntervalInMs;
+ try {
+ if (!heartbeat.getHeartbeatScheduler().awaitTermination(timeoutMs,
TimeUnit.MILLISECONDS)) {
Review Comment:
🤖 Confirmed from the PR-head source: `close()` and
`getHeartbeatWriteExecutor()` both lock `this`, and `close()` now sits in
`awaitTermination` while holding it. Worth noting the window is narrow — a tick
only contends for the monitor on the `getHeartbeatWriteExecutor()` call, so
once it's parked in `future.get` there's no contention — and the production
postCommit path goes through `stop()`, which isn't synchronized, so the fix
works as intended there. Narrowing `close()`'s `synchronized` to just the
executor teardown (after `stopHeartbeatTimers()` returns) seems like the
smallest change; otherwise a tick that loses the race lazily recreates a cached
pool that nothing ever shuts down.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java:
##########
@@ -196,15 +196,34 @@ private boolean isHeartbeatStarted(Heartbeat heartbeat) {
private void stopHeartbeatScheduler(Heartbeat heartbeat) {
log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime());
shutdownHeartbeatScheduler(heartbeat);
+ // Callers delete the heartbeat file next. A refresh landing after that
delete recreates the
+ // file, and on storage that enforces preconditions it also changes the
object generation, so a
+ // generation-matched delete is rejected (e.g. GCS 412 conditionNotMet).
+ awaitHeartbeatSchedulerTermination(heartbeat);
Review Comment:
🤖 The PR description does call this tail out explicitly ("an abandoned write
can still land after the delete") and leans on the widened catch plus the log
for it. One thing to flag on the suggested alternative: after
`future.cancel(true)`, `future.get()` returns immediately with
`CancellationException` rather than waiting for the task body, so awaiting that
future wouldn't actually close the window — you'd need a separate completion
signal from the write task. Given nothing sweeps `.heartbeat/` for completed
instants (only `stop()` and `rollbackFailedWrites` delete), the orphan is
permanent but inert, since `getInstantsToRollback` only looks at inflight
instants.
--
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]
