This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 5223271b194d2fea8250020ad3c23ac622a548ca Author: Prashant Wason <[email protected]> AuthorDate: Mon Jun 22 01:48:47 2026 -0700 fix: prevent heartbeat timer from being permanently killed by slow or delayed heartbeats (#18904) * fix: prevent heartbeat timer from being permanently killed by slow or delayed heartbeats HoodieHeartbeatClient could permanently stop generating heartbeats for an instant, causing later commits to abort with "Heartbeat for instant ... has expired" even though the writer was still alive: - The heartbeat file is written synchronously on the Timer thread. Because the timer uses scheduleAtFixedRate, a slow or hung storage write blocks the thread and freezes all subsequent heartbeats for that instant. - When a heartbeat refresh is delayed past the tolerable interval, updateHeartbeat() called Thread.currentThread().interrupt(), permanently killing the timer thread and turning a transient delay into a permanent blackout. Fix: - Perform the heartbeat file write on a bounded daemon executor (Future.get with a per-interval timeout) so a slow or hung storage call cannot block the timer thread; a timed-out write is retried on the next tick. - Remove the self-interrupt; log a warning and continue. The commit-time check HeartbeatUtils.abortIfHeartbeatExpired() remains the sole enforcement point. Add TestHoodieHeartbeatClient.testTimerSurvivesHungHeartbeatWrite. * fix: address review comments on heartbeat resilience - isHeartbeatExpired: handle a null last-heartbeat-time (the very first write can time out, leaving it unset) by falling back to the DFS read, avoiding an NPE on the unboxing comparison. - On a detected lapse, stop refreshing the heartbeat (cancel the timer) and do not advance the last-heartbeat time, so a lapsed writer still aborts at commit via HeartbeatUtils.abortIfHeartbeatExpired() and cannot resurrect a heartbeat that a concurrent cleaner (LAZY failed-writes policy) may already have acted on. The timer is cancelled cleanly rather than via Thread.interrupt(). - close(): make idempotent via a closed flag and guard executor creation after close. - Use boxed Long for heartbeatWriteTimeoutMs to match the sibling duration fields. - Raise the default hoodie.client.heartbeat.tolerable.misses from 2 to 10 so transient driver pauses (e.g. GC) or storage-latency spikes do not abort a still-healthy writer. * fix: keep heartbeat client reusable after close A write client reuses its HoodieHeartbeatClient across operations: after close(), startCommit() and acquireRollbackHeartbeatIfMultiWriter() call heartbeatClient.start() again. The previously added "already closed" guard in the executor accessor turned this valid reuse into a failure (observed in TestJavaHoodieBackedMetadata and TestHoodieJavaClientOnCopyOnWriteStorage). Remove the guard and the closed flag; close() remains idempotent via synchronization plus the null-check on the executor, so repeated/concurrent close() is still safe while reuse-after-close works (the executor is lazily re-created on the next heartbeat). * switch timer into a executor service --------- Co-authored-by: danny0405 <[email protected]> (cherry picked from commit c6f38c6b6e84212f6f9aec5c88ae271ff424e741) --- .../client/heartbeat/HoodieHeartbeatClient.java | 150 ++++++++++++++++----- .../org/apache/hudi/config/HoodieWriteConfig.java | 6 +- .../heartbeat/TestHoodieHeartbeatClient.java | 102 ++++++++++++++ 3 files changed, 224 insertions(+), 34 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 a043f73e632c..b8f2f15fdf0a 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 @@ -19,6 +19,7 @@ package org.apache.hudi.client.heartbeat; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.CustomizedThreadFactory; import org.apache.hudi.common.util.ValidationUtils; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieHeartbeatException; @@ -35,9 +36,15 @@ import java.io.IOException; import java.io.OutputStream; import java.io.Serializable; import java.util.Map; -import java.util.Timer; -import java.util.TimerTask; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static org.apache.hudi.common.heartbeat.HoodieHeartbeatUtils.getLastHeartbeatTime; @@ -58,7 +65,16 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { // heartbeat interval in millis private final Long heartbeatIntervalInMs; private final Long maxAllowableHeartbeatIntervalInMs; + // Maximum time the scheduler thread will wait for a single heartbeat file write to complete before + // abandoning it and letting the next tick retry. Bounded to one interval so that a slow/hung + // storage write cannot block the scheduler thread (and thus freeze all subsequent heartbeats). + private final Long heartbeatWriteTimeoutMs; private final Map<String, Heartbeat> instantToHeartbeatMap; + // Daemon executor used to perform the (potentially slow) storage write off the scheduler thread so the + // write can be time-bounded. A cached pool is intentional: if one write hangs, that thread is left + // parked while the next tick proceeds on a fresh thread. Lazily created and marked transient since + // this client is Serializable with a transient storage handle. + private transient ExecutorService heartbeatWriteExecutor; public HoodieHeartbeatClient(HoodieStorage storage, String basePath, Long heartbeatIntervalInMs, Integer numTolerableHeartbeatMisses) { @@ -68,9 +84,18 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { this.heartbeatFolderPath = HoodieTableMetaClient.getHeartbeatFolderPath(basePath); this.heartbeatIntervalInMs = heartbeatIntervalInMs; this.maxAllowableHeartbeatIntervalInMs = this.heartbeatIntervalInMs * numTolerableHeartbeatMisses; + this.heartbeatWriteTimeoutMs = this.heartbeatIntervalInMs; this.instantToHeartbeatMap = new ConcurrentHashMap<>(); } + private synchronized ExecutorService getHeartbeatWriteExecutor() { + if (heartbeatWriteExecutor == null) { + heartbeatWriteExecutor = + Executors.newCachedThreadPool(new CustomizedThreadFactory("heartbeat_write", true)); + } + return heartbeatWriteExecutor; + } + @Data static class Heartbeat { @@ -79,10 +104,12 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { private boolean isHeartbeatStopped = false; private Long lastHeartbeatTime; private Integer numHeartbeats = 0; - private Timer timer = new Timer(true); + private ScheduledExecutorService heartbeatScheduler = + Executors.newSingleThreadScheduledExecutor(new CustomizedThreadFactory("heartbeat_scheduler", true)); + private ScheduledFuture<?> scheduledFuture; } - class HeartbeatTask extends TimerTask { + class HeartbeatTask implements Runnable { private final String instantTime; @@ -92,7 +119,11 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { @Override public void run() { - updateHeartbeat(instantTime); + try { + updateHeartbeat(instantTime); + } catch (Exception e) { + log.error("Failed to update heartbeat for instant {}; will retry on next tick", instantTime, e); + } } } @@ -114,11 +145,11 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { newHeartbeat.setHeartbeatStarted(true); instantToHeartbeatMap.put(instantTime, newHeartbeat); // Ensure heartbeat is generated for the first time with this blocking call. - // Since timer submits the task to a thread, no guarantee when that thread will get CPU + // Since scheduler submits the task to a thread, no guarantee when that thread will get CPU // cycles to generate the first heartbeat. updateHeartbeat(instantTime); - newHeartbeat.getTimer().scheduleAtFixedRate(new HeartbeatTask(instantTime), this.heartbeatIntervalInMs, - this.heartbeatIntervalInMs); + newHeartbeat.setScheduledFuture(newHeartbeat.getHeartbeatScheduler().scheduleAtFixedRate( + new HeartbeatTask(instantTime), this.heartbeatIntervalInMs, this.heartbeatIntervalInMs, TimeUnit.MILLISECONDS)); } /** @@ -130,7 +161,7 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { public Heartbeat stop(String instantTime) throws HoodieException { Heartbeat heartbeat = instantToHeartbeatMap.remove(instantTime); if (isHeartbeatStarted(heartbeat)) { - stopHeartbeatTimer(heartbeat); + stopHeartbeatScheduler(heartbeat); HeartbeatUtils.deleteHeartbeatFile(storage, basePath, instantTime); log.info("Deleted heartbeat file for instant {}", instantTime); } @@ -138,12 +169,12 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { } /** - * Stops all timers of heartbeats started via this instance of the client. + * Stops all heartbeat schedulers started via this instance of the client. * * @throws HoodieException */ public void stopHeartbeatTimers() throws HoodieException { - instantToHeartbeatMap.values().stream().filter(this::isHeartbeatStarted).forEach(this::stopHeartbeatTimer); + instantToHeartbeatMap.values().stream().filter(this::isHeartbeatStarted).forEach(this::stopHeartbeatScheduler); } /** @@ -158,17 +189,24 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { } /** - * Stops the timer of the given heartbeat. + * Stops the scheduler of the given heartbeat. * * @param heartbeat The heartbeat to stop. */ - private void stopHeartbeatTimer(Heartbeat heartbeat) { + private void stopHeartbeatScheduler(Heartbeat heartbeat) { log.info("Stopping heartbeat for instant {}", heartbeat.getInstantTime()); - heartbeat.getTimer().cancel(); + shutdownHeartbeatScheduler(heartbeat); heartbeat.setHeartbeatStopped(true); log.info("Stopped heartbeat for instant {}", heartbeat.getInstantTime()); } + private void shutdownHeartbeatScheduler(Heartbeat heartbeat) { + if (heartbeat.getScheduledFuture() != null) { + heartbeat.getScheduledFuture().cancel(false); + } + heartbeat.getHeartbeatScheduler().shutdownNow(); + } + public static Boolean heartbeatExists(HoodieStorage storage, String basePath, String instantTime) throws IOException { StoragePath heartbeatFilePath = new StoragePath( HoodieTableMetaClient.getHeartbeatFolderPath(basePath), instantTime); @@ -178,17 +216,18 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { public boolean isHeartbeatExpired(String instantTime) throws IOException { Long currentTime = System.currentTimeMillis(); Heartbeat lastHeartbeatForWriter = instantToHeartbeatMap.get(instantTime); - if (lastHeartbeatForWriter == null) { - log.info("Heartbeat not found in internal map, falling back to reading from DFS"); - long lastHeartbeatForWriterTime = getLastHeartbeatTime(this.storage, basePath, instantTime); - lastHeartbeatForWriter = new Heartbeat(); - lastHeartbeatForWriter.setLastHeartbeatTime(lastHeartbeatForWriterTime); - lastHeartbeatForWriter.setInstantTime(instantTime); - lastHeartbeatForWriter.getTimer().cancel(); + Long lastHeartbeatTime = lastHeartbeatForWriter == null ? null : lastHeartbeatForWriter.getLastHeartbeatTime(); + // lastHeartbeatTime can be null when the heartbeat is not in the internal map, or when it is in the + // map but no heartbeat has been generated yet (e.g. the first write timed out). In both cases fall + // back to reading the last heartbeat time from DFS (returns 0 if no heartbeat file exists, which is + // correctly treated as expired). + if (lastHeartbeatTime == null) { + log.info("Heartbeat time not available in internal map, falling back to reading from DFS"); + lastHeartbeatTime = getLastHeartbeatTime(this.storage, basePath, instantTime); } - if (currentTime - lastHeartbeatForWriter.getLastHeartbeatTime() > this.maxAllowableHeartbeatIntervalInMs) { + if (currentTime - lastHeartbeatTime > this.maxAllowableHeartbeatIntervalInMs) { log.warn("Heartbeat expired, currentTime = {}, last heartbeat = {}, heartbeat interval = {}", currentTime, - lastHeartbeatForWriter, this.heartbeatIntervalInMs); + lastHeartbeatTime, this.heartbeatIntervalInMs); return true; } return false; @@ -197,20 +236,31 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { private void updateHeartbeat(String instantTime) throws HoodieHeartbeatException { try { Long newHeartbeatTime = System.currentTimeMillis(); - OutputStream outputStream = - this.storage.create( - new StoragePath(heartbeatFolderPath, instantTime), true); - outputStream.close(); + writeHeartbeatFile(instantTime); Heartbeat heartbeat = instantToHeartbeatMap.get(instantTime); if (heartbeat.getLastHeartbeatTime() != null && isHeartbeatExpired(instantTime)) { - log.error("Aborting, missed generating heartbeat within allowable interval {} ms", this.maxAllowableHeartbeatIntervalInMs); - // Since TimerTask allows only java.lang.Runnable, cannot throw an exception and bubble to the caller thread, hence - // explicitly interrupting the timer thread. - Thread.currentThread().interrupt(); + // 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 + // and the writer aborts at commit time via HeartbeatUtils.abortIfHeartbeatExpired(). We must not + // keep refreshing here: a concurrent process (e.g. an async cleaner under LAZY failed-writes + // policy) may already have started rolling back this instant once it observed the expiry, and + // resurrecting the heartbeat could let this writer commit on top of rolled-back files. + // The scheduler is cancelled cleanly rather than via Thread.interrupt(), which would permanently + // kill the scheduler thread (turning a transient delay into a permanent blackout on the first miss). + log.error("Missed generating heartbeat for instant {} within allowable interval {} ms; stopping heartbeat refresh", + instantTime, this.maxAllowableHeartbeatIntervalInMs); + shutdownHeartbeatScheduler(heartbeat); + return; } heartbeat.setInstantTime(instantTime); heartbeat.setLastHeartbeatTime(newHeartbeatTime); heartbeat.setNumHeartbeats(heartbeat.getNumHeartbeats() + 1); + } catch (TimeoutException te) { + // The storage write did not complete within the bounded window. Do not advance the last heartbeat + // time (the write is unconfirmed); the next scheduled tick will retry on a fresh executor thread. + // Crucially, the scheduler thread is freed instead of being blocked by a hung storage call. + 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) { @@ -221,13 +271,49 @@ public class HoodieHeartbeatClient implements AutoCloseable, Serializable { } } + /** + * Writes the heartbeat file for the given instant on a dedicated daemon executor, bounded by + * {@link #heartbeatWriteTimeoutMs}. Performing the storage write off the scheduler thread (and with a + * timeout) ensures that a slow or hung storage call cannot block the scheduler thread and freeze all + * subsequent heartbeats for this instant. + */ + private void writeHeartbeatFile(String instantTime) throws IOException, TimeoutException { + Future<Void> future = getHeartbeatWriteExecutor().submit(() -> { + try (OutputStream outputStream = + this.storage.create(new StoragePath(heartbeatFolderPath, instantTime), true)) { + // create + close confirms the heartbeat file write landed on storage. + } + return null; + }); + try { + future.get(heartbeatWriteTimeoutMs, TimeUnit.MILLISECONDS); + } catch (TimeoutException te) { + future.cancel(true); + throw te; + } catch (InterruptedException ie) { + future.cancel(true); + Thread.currentThread().interrupt(); + throw new HoodieHeartbeatException("Interrupted while writing heartbeat for instant " + instantTime, ie); + } catch (ExecutionException ee) { + Throwable cause = ee.getCause(); + if (cause instanceof IOException) { + throw (IOException) cause; + } + throw new HoodieHeartbeatException("Failed to write heartbeat for instant " + instantTime, cause); + } + } + public Heartbeat getHeartbeat(String instantTime) { return this.instantToHeartbeatMap.get(instantTime); } @Override - public void close() { + public synchronized void close() { this.stopHeartbeatTimers(); this.instantToHeartbeatMap.clear(); + if (heartbeatWriteExecutor != null) { + heartbeatWriteExecutor.shutdownNow(); + heartbeatWriteExecutor = null; + } } } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index a663c63dba78..1ba577faabf1 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -680,9 +680,11 @@ public class HoodieWriteConfig extends HoodieConfig { public static final ConfigProperty<Integer> CLIENT_HEARTBEAT_NUM_TOLERABLE_MISSES = ConfigProperty .key("hoodie.client.heartbeat.tolerable.misses") - .defaultValue(2) + .defaultValue(10) .markAdvanced() - .withDocumentation("Number of heartbeat misses, before a writer is deemed not alive and all pending writes are aborted."); + .withDocumentation("Number of heartbeat misses, before a writer is deemed not alive and all pending writes are aborted. " + + "A higher value tolerates transient driver pauses (e.g. GC) or storage-latency spikes that would otherwise " + + "delay a heartbeat and cause a still-healthy writer's commit to be aborted."); public static final ConfigProperty<Boolean> CLUSTERING_BLOCK_FOR_PENDING_INGESTION = ConfigProperty .key("hoodie.clustering.fail.on.pending.ingestion.during.conflict.resolution") 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 c7ea5fa87bbd..5feb79ee83bc 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 @@ -21,12 +21,18 @@ package org.apache.hudi.client.heartbeat; import org.apache.hudi.common.testutils.HoodieCommonTestHarness; import org.apache.hudi.storage.StoragePath; import org.apache.hudi.storage.StoragePathInfo; +import org.apache.hudi.storage.hadoop.HoodieHadoopStorage; +import org.apache.hadoop.fs.FileSystem; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.io.OutputStream; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static java.util.concurrent.TimeUnit.SECONDS; import static org.awaitility.Awaitility.await; @@ -113,4 +119,100 @@ public class TestHoodieHeartbeatClient extends HoodieCommonTestHarness { assertFalse(hoodieHeartbeatClient.isHeartbeatExpired(instantTime1)); assertTrue(hoodieHeartbeatClient.getHeartbeat(instantTime1).isHeartbeatStopped()); } + + /** + * Regression test for the heartbeat-expiry incident: a single slow/hung storage write must not + * block (freeze) the heartbeat scheduler thread. The first heartbeat write blocks (simulating a hung + * cloud-storage call); we assert the scheduler keeps producing heartbeats on fresh threads once that + * write times out, proving the scheduler thread was not blocked by the synchronous storage call (#1). + * A high tolerable-misses is used so that recovery after the blocked write does not itself trip the + * expiry path (which intentionally stops refresh on a genuine lapse). + */ + @Test + public void testSlowHeartbeatWriteDoesNotBlockScheduler() { + CountDownLatch releaseFirstWrite = new CountDownLatch(1); + SlowCreateStorage slowStorage = + new SlowCreateStorage((FileSystem) metaClient.getStorage().getFileSystem(), releaseFirstWrite); + // interval 1s, write timeout = 1s; high tolerable-misses so the ~1s recovery gap stays well within + // the allowable window and the scheduler keeps beating rather than treating it as a lapse. + HoodieHeartbeatClient hoodieHeartbeatClient = + new HoodieHeartbeatClient(slowStorage, metaClient.getBasePath().toString(), + heartBeatInterval, 10); + try { + hoodieHeartbeatClient.start(instantTime1); + // Despite the first write hanging, the scheduler must keep generating heartbeats on fresh threads. + await().atMost(15, SECONDS) + .until(() -> hoodieHeartbeatClient.getHeartbeat(instantTime1).getNumHeartbeats() >= 2); + } finally { + releaseFirstWrite.countDown(); + hoodieHeartbeatClient.close(); + } + } + + @Test + public void testScheduledHeartbeatRetriesAfterWriteFailure() { + FailOnceAfterInitialCreateStorage storage = + new FailOnceAfterInitialCreateStorage((FileSystem) metaClient.getStorage().getFileSystem()); + HoodieHeartbeatClient hoodieHeartbeatClient = + new HoodieHeartbeatClient(storage, metaClient.getBasePath().toString(), heartBeatInterval, 10); + try { + hoodieHeartbeatClient.start(instantTime1); + await().atMost(10, SECONDS).until(storage::hasInjectedFailure); + await().atMost(10, SECONDS) + .until(() -> hoodieHeartbeatClient.getHeartbeat(instantTime1).getNumHeartbeats() >= 2); + } finally { + hoodieHeartbeatClient.close(); + } + } + + /** + * A storage wrapper whose first {@code create()} call blocks until released, simulating a hung + * storage write. All subsequent calls delegate normally. + */ + private static class SlowCreateStorage extends HoodieHadoopStorage { + + private final AtomicBoolean firstCall = new AtomicBoolean(true); + private final CountDownLatch releaseFirstWrite; + + SlowCreateStorage(FileSystem fs, CountDownLatch releaseFirstWrite) { + super(fs); + this.releaseFirstWrite = releaseFirstWrite; + } + + @Override + public OutputStream create(StoragePath path, boolean overwrite) throws IOException { + if (firstCall.getAndSet(false)) { + try { + releaseFirstWrite.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while simulating a hung heartbeat write", e); + } + } + return super.create(path, overwrite); + } + } + + private static class FailOnceAfterInitialCreateStorage extends HoodieHadoopStorage { + + private final AtomicInteger createCalls = new AtomicInteger(0); + private final AtomicBoolean injectedFailure = new AtomicBoolean(false); + + FailOnceAfterInitialCreateStorage(FileSystem fs) { + super(fs); + } + + @Override + public OutputStream create(StoragePath path, boolean overwrite) throws IOException { + int currentCall = createCalls.incrementAndGet(); + if (currentCall == 2 && injectedFailure.compareAndSet(false, true)) { + throw new IOException("Injected scheduled heartbeat write failure"); + } + return super.create(path, overwrite); + } + + private boolean hasInjectedFailure() { + return injectedFailure.get(); + } + } }
