hudi-agent commented on code in PR #19867:
URL: https://github.com/apache/hudi/pull/19867#discussion_r3956292898
##########
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:
π€ `close()` is `synchronized` and so is `getHeartbeatWriteExecutor()`, both
on `this`. If a tick has just entered `updateHeartbeat` when `close()` starts,
it blocks on the monitor while `close()` sits here in `awaitTermination`
holding it β so `close()` stalls for the full timeout (2Γinterval, ~120s at
defaults), and after it nulls `heartbeatWriteExecutor` the tick lazily creates
a fresh pool. Could the await be done outside the monitor, or the executor
guarded by a separate lock / created eagerly?
<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);
Review Comment:
π€ When the bounded write itself times out, `future.cancel(true)` interrupts
the write thread but a storage upload already on the wire may still land; the
scheduler thread returns, this await succeeds, and the delete runs before that
late create. Is that residual window intended to be covered only by the new
`HoodieIOException` catch (leaving an orphan heartbeat file), or would it be
worth also awaiting the outstanding write future here?
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/heartbeat/TestHoodieHeartbeatClient.java:
##########
@@ -165,6 +166,91 @@ public void
testScheduledHeartbeatRetriesAfterWriteFailure() {
}
}
+ /**
+ * stop() must not delete the heartbeat file while a scheduled refresh is
still writing it. A
+ * refresh landing after the delete recreates the file, and on storage that
enforces preconditions
+ * it changes the object generation so a generation-matched delete is
rejected (e.g. GCS 412).
+ */
+ @Test
+ public void testStopWaitsForInFlightHeartbeatRefresh() throws Exception {
+ CountDownLatch refreshEntered = new CountDownLatch(1);
+ CountDownLatch releaseRefresh = new CountDownLatch(1);
+ OrderRecordingStorage storage = new OrderRecordingStorage(
+ (FileSystem) metaClient.getStorage().getFileSystem(), refreshEntered,
releaseRefresh);
+ HoodieHeartbeatClient client = new HoodieHeartbeatClient(
+ storage, metaClient.getBasePath().toString(), heartBeatInterval,
numTolerableMisses);
Review Comment:
π€ With `heartBeatInterval = 1000L` the write timeout is also 1s. If thread
start + Awaitility's default 100ms poll delay/interval push the gap between
`refreshEntered` and `releaseRefresh.countDown()` past ~1s on a loaded CI box,
`future.get` times out, `awaitTermination` returns immediately, the delete runs
before the gated create lands, and the last-event assertion fails spuriously.
Would a larger interval for this test (e.g. 5000L like
`testNumHeartbeatsGenerated`) be safer?
<sub><i>β οΈ AI-generated; verify before applying. React π/π to flag
quality.</i></sub>
--
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]
