This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new bd38321d4370 fix(heartbeat): don't fail a durable commit when
heartbeat cleanup races a refresh (#19867)
bd38321d4370 is described below
commit bd38321d437052c869b47355001a84c8d9c11ebd
Author: vamsikarnika <[email protected]>
AuthorDate: Tue Sep 22 03:39:41 2026 +0530
fix(heartbeat): don't fail a durable commit when heartbeat cleanup races a
refresh (#19867)
stop() cancelled the heartbeat scheduler with shutdownNow() and deleted the
heartbeat file without waiting for an in-flight refresh. The refresh
could land
after the delete, recreating the file and bumping the object generation,
so a
generation-matched delete was rejected (GCS 412 conditionNotMet).
HoodieHadoopStorage.delete throws unchecked HoodieIOException when the
object
still exists after a refused delete, which escaped the IOException-only
catch in
deleteHeartbeatFile, propagated out of postCommit and aborted commitStats
-- after
commit() had already made the write durable. The caller saw a successful
commit
reported as a failure.
- HoodieHeartbeatClient.stopHeartbeatScheduler: shutdown() plus bounded
awaitTermination() so an in-flight refresh completes before the delete.
Not
placed in shutdownHeartbeatScheduler, which the missed-refresh path
calls from
the scheduler thread itself and must not await itself.
- WriterHeartbeatUtils.deleteHeartbeatFile: also catch HoodieIOException.
The
commit is durable by postCommit and every caller discards the return
value, so
failed cleanup of a transient lease marker must not be fatal. The
log.error is
kept, so a leaked file stays visible.
- updateHeartbeat: guard against a null map entry, since stop() removes
the entry
before stopping the scheduler and a refresh can complete inside the new
await
window.
---
.../client/heartbeat/HoodieHeartbeatClient.java | 37 ++++-
.../client/heartbeat/WriterHeartbeatUtils.java | 7 +-
.../heartbeat/TestHoodieHeartbeatClient.java | 152 +++++++++++++++++++++
3 files changed, 190 insertions(+), 6 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
index beeae36a4246..858c36db3b58 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/HoodieHeartbeatClient.java
@@ -196,15 +196,34 @@ public class HoodieHeartbeatClient implements
AutoCloseable, Serializable {
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)) {
+ log.warn("Timed out after {} ms awaiting an in-flight heartbeat
refresh for instant {}",
+ timeoutMs, heartbeat.getInstantTime());
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted awaiting heartbeat scheduler termination for
instant {}", heartbeat.getInstantTime());
+ }
}
public static Boolean heartbeatExists(HoodieStorage storage, String
basePath, String instantTime) throws IOException {
@@ -238,6 +257,10 @@ public class HoodieHeartbeatClient implements
AutoCloseable, Serializable {
Long newHeartbeatTime = System.currentTimeMillis();
writeHeartbeatFile(instantTime);
Heartbeat heartbeat = instantToHeartbeatMap.get(instantTime);
+ if (heartbeat == null) {
+ // stop() removed the entry while this refresh was in flight.
+ return;
+ }
if (heartbeat.getLastHeartbeatTime() != null &&
isHeartbeatExpired(instantTime)) {
// A previous refresh was delayed past the tolerable interval. Stop
refreshing this heartbeat
// (cancel the scheduler) and do NOT advance the last heartbeat time,
so the heartbeat stays expired
@@ -262,8 +285,8 @@ public class HoodieHeartbeatClient implements
AutoCloseable, Serializable {
log.warn("Heartbeat file write for instant {} did not complete within {}
ms; will retry on next tick",
instantTime, this.heartbeatWriteTimeoutMs);
} catch (IOException io) {
- boolean isHeartbeatStopped =
instantToHeartbeatMap.get(instantTime).isHeartbeatStopped();
- if (isHeartbeatStopped) {
+ Heartbeat heartbeat = instantToHeartbeatMap.get(instantTime);
+ if (heartbeat == null || heartbeat.isHeartbeatStopped()) {
log.info("update heart beat failed, because the instant time {} was
stopped", instantTime);
return;
}
@@ -308,9 +331,15 @@ public class HoodieHeartbeatClient implements
AutoCloseable, Serializable {
}
@Override
- public synchronized void close() {
+ public void close() {
+ // Not synchronized: stopHeartbeatTimers() now awaits in-flight refreshes,
and holding the monitor
+ // across that would block a refresh needing getHeartbeatWriteExecutor()
until the await expires.
this.stopHeartbeatTimers();
this.instantToHeartbeatMap.clear();
+ shutdownHeartbeatWriteExecutor();
+ }
+
+ private synchronized void shutdownHeartbeatWriteExecutor() {
if (heartbeatWriteExecutor != null) {
heartbeatWriteExecutor.shutdownNow();
heartbeatWriteExecutor = null;
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/WriterHeartbeatUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/WriterHeartbeatUtils.java
index 9763142628e4..1cd61f4cab36 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/WriterHeartbeatUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/heartbeat/WriterHeartbeatUtils.java
@@ -22,6 +22,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
@@ -58,8 +59,10 @@ public class WriterHeartbeatUtils {
} 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) {
+ // HoodieStorage.deleteFile throws HoodieIOException when the object
still exists after a
+ // rejected delete. Never fatal: the commit is already durable by
postCommit.
+ log.error("Unable to delete heartbeat for instant {}", instantTime, e);
}
return deleted;
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/heartbeat/TestHoodieHeartbeatClient.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/heartbeat/TestHoodieHeartbeatClient.java
index 2df16a1c90c9..eb3f069aa655 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/heartbeat/TestHoodieHeartbeatClient.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/heartbeat/TestHoodieHeartbeatClient.java
@@ -19,6 +19,7 @@
package org.apache.hudi.client.heartbeat;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
@@ -30,6 +31,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.io.OutputStream;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -165,6 +167,156 @@ public class TestHoodieHeartbeatClient extends
HoodieCommonTestHarness {
}
}
+ /**
+ * 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 {
+ // A longer interval also widens the bounded-write timeout, so a slow CI
box cannot let the write
+ // time out before the test releases it, which would let the delete run
first.
+ long interval = 5000L;
+ 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(), interval,
numTolerableMisses);
+ try {
+ client.start(instantTime1);
+ assertTrue(refreshEntered.await(20, SECONDS), "Scheduled heartbeat
refresh never started");
+
+ Thread stopper = new Thread(() -> client.stop(instantTime1));
+ stopper.start();
+ // Without awaiting termination, stop() runs straight through and never
parks here.
+ await().atMost(20, SECONDS).until(() -> stopper.getState() ==
Thread.State.TIMED_WAITING
+ || stopper.getState() == Thread.State.WAITING
+ || stopper.getState() == Thread.State.BLOCKED);
+
+ releaseRefresh.countDown();
+ stopper.join(SECONDS.toMillis(20));
+ assertFalse(stopper.isAlive(), "stop() did not complete after the
refresh was released");
+
+ assertTrue(storage.events().contains("delete"), "stop() never deleted
the heartbeat file");
+ assertEquals("delete", storage.events().get(storage.events().size() - 1),
+ "No heartbeat refresh may follow the delete; events were " +
storage.events());
+ } finally {
+ releaseRefresh.countDown();
+ client.close();
+ }
+ }
+
+ /**
+ * A rejected heartbeat delete must never propagate.
HoodieStorage.deleteFile throws
+ * HoodieIOException (unchecked) when the object still exists after the
delete was refused, which is
+ * what a generation-matched delete does on storage that enforces
preconditions. Callers reach this
+ * from postCommit, where the commit is already durable.
+ */
+ @Test
+ public void testDeleteHeartbeatFileSwallowsHoodieIOException() {
+ ThrowOnDeleteStorage storage =
+ new ThrowOnDeleteStorage((FileSystem)
metaClient.getStorage().getFileSystem());
+ assertFalse(WriterHeartbeatUtils.deleteHeartbeatFile(storage, basePath,
instantTime1),
+ "A refused heartbeat delete must be reported via the return value,
never thrown");
+ }
+
+ /** Interrupting a thread parked in stop() must re-assert the interrupt
rather than swallow it. */
+ @Test
+ public void testStopReassertsInterruptWhileAwaiting() throws Exception {
+ // A longer interval widens the bounded-write window, so the await is
still parked when interrupted.
+ long interval = 5000L;
+ 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(), interval,
numTolerableMisses);
+ try {
+ client.start(instantTime1);
+ assertTrue(refreshEntered.await(20, SECONDS), "Scheduled heartbeat
refresh never started");
+
+ AtomicBoolean interruptPreserved = new AtomicBoolean();
+ Thread stopper = new Thread(() -> {
+ client.stop(instantTime1);
+ interruptPreserved.set(Thread.currentThread().isInterrupted());
+ });
+ stopper.start();
+ await().atMost(20, SECONDS).until(() -> stopper.getState() ==
Thread.State.TIMED_WAITING
+ || stopper.getState() == Thread.State.WAITING
+ || stopper.getState() == Thread.State.BLOCKED);
+ stopper.interrupt();
+ stopper.join(SECONDS.toMillis(20));
+
+ assertFalse(stopper.isAlive(), "stop() did not return after being
interrupted");
+ assertTrue(interruptPreserved.get(), "stop() swallowed the interrupt");
+ } finally {
+ releaseRefresh.countDown();
+ client.close();
+ }
+ }
+
+ /** Refuses every delete with HoodieIOException, as a precondition-enforcing
store does. */
+ private static class ThrowOnDeleteStorage extends HoodieHadoopStorage {
+
+ ThrowOnDeleteStorage(FileSystem fs) {
+ super(fs);
+ }
+
+ @Override
+ public boolean deleteFile(StoragePath path) {
+ throw new HoodieIOException("Failed to delete invalid data file: " +
path);
+ }
+ }
+
+ /**
+ * Records create/delete order and blocks the first scheduled refresh until
released. The block
+ * ignores interruption, as a storage write already in flight would.
+ */
+ private static class OrderRecordingStorage extends HoodieHadoopStorage {
+
+ private final List<String> events = new CopyOnWriteArrayList<>();
+ private final AtomicBoolean gated = new AtomicBoolean(false);
+ private final CountDownLatch refreshEntered;
+ private final CountDownLatch releaseRefresh;
+
+ OrderRecordingStorage(FileSystem fs, CountDownLatch refreshEntered,
CountDownLatch releaseRefresh) {
+ super(fs);
+ this.refreshEntered = refreshEntered;
+ this.releaseRefresh = releaseRefresh;
+ }
+
+ List<String> events() {
+ return events;
+ }
+
+ @Override
+ public OutputStream create(StoragePath path, boolean overwrite) throws
IOException {
+ // The first create is start()'s synchronous beat; gate only the first
scheduled refresh.
+ if (!events.isEmpty() && gated.compareAndSet(false, true)) {
+ refreshEntered.countDown();
+ boolean released = false;
+ while (!released) {
+ try {
+ releaseRefresh.await();
+ released = true;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ OutputStream stream = super.create(path, overwrite);
+ events.add("create");
+ return stream;
+ }
+
+ @Override
+ public boolean deleteFile(StoragePath path) throws IOException {
+ events.add("delete");
+ return super.deleteFile(path);
+ }
+ }
+
/**
* A storage wrapper whose first {@code create()} call blocks until
released, simulating a hung
* storage write. All subsequent calls delegate normally.
