From ec0ea3b8cc58b8a99a8e45c71a4cec7bf4c64c7f Mon Sep 17 00:00:00 2001
From: Praveen Gajulapalli <13733716+pkgajulapalli@users.noreply.github.com>
Date: Mon, 10 Aug 2026 16:44:51 +0530
Subject: [PATCH] fix(lock): name the cause on FAILED_TO_RELEASE in
 StorageBasedLockProvider

Three distinct failures in StorageBasedLockProvider#unlock() threw a
byte-identical HoodieLockException message, so logs could not tell them
apart. All three also share a single updateLockReleaseFailureMetric
counter, leaving no way to attribute a release failure to a cause.

Each throw now names its cause, and each logs the context needed to act
on it:

- HEARTBEAT_STOP_FAILED: the heartbeat task would not stop, so the lock
  is deliberately left un-expired (the task could still renew it after
  we return). Logs the interrupted flag to separate the two sub-cases in
  LockProviderHeartbeatManager#stopHeartbeat.
- INTERRUPTED_DURING_THROTTLE_BACKOFF: interrupted mid-backoff. Now also
  passes the InterruptedException so the stack trace survives.
- THROTTLE_RETRIES_EXHAUSTED vs EXPIRE_WRITE_FAILED: distinguishes an
  exhausted retry budget against a storage rate limit (e.g. the GCS
  1-write/sec per-object limit) from a terminal UNKNOWN_ERROR /
  ACQUIRED_BY_OTHERS outcome.

The four cause strings are declared as constants next to the other lock
tunables, so the full set is visible in one place and both the call sites
and the test assertions reference them rather than raw literals.

On ACQUIRED_BY_OTHERS, also log how long ago our lease should have
ended. A positive value means we overran our own lease, pointing at a
starved heartbeat (long GC, thread-pool starvation); a negative value
means the lease had not elapsed by our clock, pointing at clock skew
between nodes instead. Those two causes are indistinguishable today and
call for different fixes.

Every new message carries lockFilePath, so a lock left dangling in
storage can be joined back to the writer that failed to release it.

Behaviour is unchanged: control flow is untouched, and each edit either
adds a logger.error call or appends ", cause <LABEL>" to an existing
exception message. Metrics are unchanged.

Adds a test for the interrupted-during-backoff path, which had no
coverage, and tightens the three existing FAILED_TO_RELEASE assertions
to pin the specific cause label.
---
 .../lock/StorageBasedLockProvider.java        | 56 ++++++++++++++++++-
 .../lock/TestStorageBasedLockProvider.java    | 41 +++++++++++++-
 2 files changed, 93 insertions(+), 4 deletions(-)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
index 974ea01aee7af..632bd99687a98 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/StorageBasedLockProvider.java
@@ -89,6 +89,21 @@ public class StorageBasedLockProvider implements LockProvider<StorageLockFile> {
   @VisibleForTesting
   static final long THROTTLE_INITIAL_RETRY_DELAY_SECONDS = 1;
 
+  // The full set of causes reported alongside FAILED_TO_RELEASE. Several distinct failures all
+  // surface as that one lock state, so the cause is what tells them apart in production logs.
+  // The heartbeat task would not stop, so the lock is deliberately left un-expired.
+  @VisibleForTesting
+  static final String CAUSE_HEARTBEAT_STOP_FAILED = "HEARTBEAT_STOP_FAILED";
+  // Interrupted while backing off between throttled expire-write attempts.
+  @VisibleForTesting
+  static final String CAUSE_INTERRUPTED_DURING_THROTTLE_BACKOFF = "INTERRUPTED_DURING_THROTTLE_BACKOFF";
+  // Every expire-write attempt was throttled by storage; the retry budget ran out.
+  @VisibleForTesting
+  static final String CAUSE_THROTTLE_RETRIES_EXHAUSTED = "THROTTLE_RETRIES_EXHAUSTED";
+  // Terminal expire-write outcome: UNKNOWN_ERROR or ACQUIRED_BY_OTHERS.
+  @VisibleForTesting
+  static final String CAUSE_EXPIRE_WRITE_FAILED = "EXPIRE_WRITE_FAILED";
+
   // Use for testing
   private final Logger logger;
 
@@ -458,8 +473,15 @@ public void unlock() {
       if (heartbeatManager.hasActiveHeartbeat()) {
         logger.debug("Owner {}: Gracefully shutting down heartbeat.", ownerId);
         if (!heartbeatManager.stopHeartbeat(true)) {
+          // The heartbeat task would not stop, so we must not expire the lock: the task could
+          // still renew it after we returned. See LockProviderHeartbeatManager#stopHeartbeat for
+          // which of the two sub-cases (interrupted vs. still-inflight) was logged.
+          logger.error("Owner {}: Cannot release lock {} - heartbeat failed to stop, so the lock is "
+                  + "left un-expired and will be reclaimed only after its lease elapses. "
+                  + "interrupted={}", ownerId, lockFilePath, Thread.currentThread().isInterrupted());
           hoodieLockMetrics.ifPresent(HoodieLockMetrics::updateLockReleaseFailureMetric);
-          throw new HoodieLockException(generateLockStateMessage(FAILED_TO_RELEASE));
+          throw new HoodieLockException(
+              generateLockStateMessage(FAILED_TO_RELEASE, CAUSE_HEARTBEAT_STOP_FAILED));
         }
       }
 
@@ -483,8 +505,12 @@ public void unlock() {
         // Re-set the interrupt flag and abandon the retry — an interrupted thread shouldn't keep
         // doing work. The caller will see FAILED_TO_RELEASE below.
         Thread.currentThread().interrupt();
+        logger.error("Owner {}: Cannot release lock {} - interrupted while backing off after "
+                + "throttled expire write (attempt {}/{}); lock left un-expired.",
+            ownerId, lockFilePath, attempt, THROTTLE_MAX_RETRIES, ie);
         hoodieLockMetrics.ifPresent(HoodieLockMetrics::updateLockReleaseFailureMetric);
-        throw new HoodieLockException(generateLockStateMessage(FAILED_TO_RELEASE));
+        throw new HoodieLockException(
+            generateLockStateMessage(FAILED_TO_RELEASE, CAUSE_INTERRUPTED_DURING_THROTTLE_BACKOFF));
       }
       synchronized (this) {
         // Bail out if the lock was either cleared by another path (e.g. shutdown hook,
@@ -498,8 +524,16 @@ public void unlock() {
     }
 
     if (expireResult != ExpireLockResult.SUCCESS) {
+      // THROTTLED here means the retries above were exhausted; FAILED means tryExpireCurrentLock
+      // already logged the specific storage outcome (UNKNOWN_ERROR vs ACQUIRED_BY_OTHERS).
+      String cause = expireResult == ExpireLockResult.THROTTLED
+          ? CAUSE_THROTTLE_RETRIES_EXHAUSTED
+          : CAUSE_EXPIRE_WRITE_FAILED;
+      logger.error("Owner {}: Cannot release lock {} - expire write ended as {} after {} throttle "
+              + "retries; lock left un-expired and will dangle until its lease elapses.",
+          ownerId, lockFilePath, expireResult, THROTTLE_MAX_RETRIES);
       hoodieLockMetrics.ifPresent(HoodieLockMetrics::updateLockReleaseFailureMetric);
-      throw new HoodieLockException(generateLockStateMessage(FAILED_TO_RELEASE));
+      throw new HoodieLockException(generateLockStateMessage(FAILED_TO_RELEASE, cause));
     }
   }
 
@@ -566,6 +600,13 @@ synchronized ExpireLockResult tryExpireCurrentLock(boolean fromShutdownHook) {
         return ExpireLockResult.SUCCESS;
       case ACQUIRED_BY_OTHERS:
         // Lock was acquired by others, indicating heartbeat failure during lock hold period.
+        // Log how long ago our lease should have ended: a positive value means we overran it,
+        // which distinguishes a starved heartbeat from a premature steal by a skewed clock.
+        logger.error("Owner {}: Lock {} was acquired by another owner before we could expire it, "
+                + "indicating heartbeat failure during the hold period. Our lease validUntil was "
+                + "{} ms ago (negative means the lease had not yet elapsed by our clock, which "
+                + "points at clock skew rather than a stalled heartbeat).",
+            ownerId, lockFilePath, getCurrentEpochMs() - this.getLock().getValidUntilMs());
         logErrorLockState(FAILED_TO_RELEASE, "lock was acquired by others, indicating heartbeat failure.");
         setLock(null);
         hoodieLockMetrics.ifPresent(HoodieLockMetrics::updateLockAcquiredByOthersErrorMetric);
@@ -678,6 +719,15 @@ private String generateLockStateMessage(LockState state) {
         state.toString());
   }
 
+  /**
+   * Same as {@link #generateLockStateMessage(LockState)}, but names the specific cause.
+   * Several distinct failures all surface as FAILED_TO_RELEASE; without the cause the
+   * exception alone cannot tell them apart in production logs.
+   */
+  private String generateLockStateMessage(LockState state, String cause) {
+    return String.format("%s, cause %s", generateLockStateMessage(state), cause);
+  }
+
   private static final String LOCK_STATE_LOGGER_MSG = "Owner {}: Lock file path {}, Thread {}, Storage based lock state {}";
   private static final String LOCK_STATE_LOGGER_MSG_WITH_INFO = "Owner {}: Lock file path {}, Thread {}, Storage based lock state {}, {}";
 
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestStorageBasedLockProvider.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestStorageBasedLockProvider.java
index 85c21c129d0fb..5a9775589dca8 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestStorageBasedLockProvider.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/lock/TestStorageBasedLockProvider.java
@@ -381,7 +381,10 @@ void testUnlockFailsToStopHeartbeat() {
     assertTrue(lockProvider.tryLock());
     when(mockHeartbeatManager.stopHeartbeat(true)).thenReturn(false);
     when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(true);
-    assertThrows(HoodieLockException.class, () -> lockProvider.unlock());
+    HoodieLockException exception = assertThrows(HoodieLockException.class, () -> lockProvider.unlock());
+    assertTrue(exception.getMessage().contains("FAILED_TO_RELEASE"));
+    // The cause must distinguish this from the other FAILED_TO_RELEASE paths.
+    assertTrue(exception.getMessage().contains(StorageBasedLockProvider.CAUSE_HEARTBEAT_STOP_FAILED), exception.getMessage());
     when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(false);
   }
 
@@ -404,6 +407,8 @@ void testUnlockThrowsExceptionWhenLockAcquiredByOthers() {
 
     HoodieLockException exception = assertThrows(HoodieLockException.class, () -> lockProvider.unlock());
     assertTrue(exception.getMessage().contains("FAILED_TO_RELEASE"));
+    // A steal is a terminal expire-write failure, not an exhausted throttle budget.
+    assertTrue(exception.getMessage().contains(StorageBasedLockProvider.CAUSE_EXPIRE_WRITE_FAILED), exception.getMessage());
     when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(false);
   }
 
@@ -495,6 +500,8 @@ void testUnlockThrowsExceptionWhenStillThrottledAfterAllRetries() throws Interru
 
     HoodieLockException exception = assertThrows(HoodieLockException.class, () -> lockProvider.unlock());
     assertTrue(exception.getMessage().contains("FAILED_TO_RELEASE"));
+    // Exhausting the retry budget must be distinguishable from a hard expire-write failure.
+    assertTrue(exception.getMessage().contains(StorageBasedLockProvider.CAUSE_THROTTLE_RETRIES_EXHAUSTED), exception.getMessage());
     // 1 initial attempt + THROTTLE_MAX_RETRIES retries.
     verify(mockLockService, times(1 + StorageBasedLockProvider.THROTTLE_MAX_RETRIES))
         .tryUpsertLockFile(any(), eq(Option.of(realLockFile)));
@@ -506,6 +513,38 @@ void testUnlockThrowsExceptionWhenStillThrottledAfterAllRetries() throws Interru
     when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(false);
   }
 
+  @Test
+  void testUnlockThrowsExceptionWhenInterruptedDuringThrottleBackoff() throws InterruptedException {
+    // The first expire attempt is THROTTLED, then the backoff sleep is interrupted. unlock()
+    // must abandon the retry, re-set the interrupt flag, and report the interruption as the
+    // cause rather than an exhausted retry budget.
+    when(mockLockService.readCurrentLockFile()).thenReturn(Pair.of(LockGetResult.NOT_EXISTS, Option.empty()));
+    StorageLockData data = new StorageLockData(false, System.currentTimeMillis() + DEFAULT_LOCK_VALIDITY_MS, ownerId);
+    StorageLockFile realLockFile = new StorageLockFile(data, "v1");
+    when(mockLockService.tryUpsertLockFile(any(), eq(Option.empty())))
+        .thenReturn(Pair.of(LockUpsertResult.SUCCESS, Option.of(realLockFile)));
+    when(mockHeartbeatManager.startHeartbeatForThread(any())).thenReturn(true);
+    assertTrue(lockProvider.tryLock());
+
+    when(mockHeartbeatManager.stopHeartbeat(true)).thenReturn(true);
+    when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(true).thenReturn(false);
+    when(mockLockService.tryUpsertLockFile(any(), eq(Option.of(realLockFile))))
+        .thenReturn(Pair.of(LockUpsertResult.THROTTLED, Option.empty()));
+    doThrow(new InterruptedException("interrupted while backing off"))
+        .when(lockProvider).sleepForThrottleRetry(anyLong());
+
+    HoodieLockException exception = assertThrows(HoodieLockException.class, () -> lockProvider.unlock());
+    assertTrue(exception.getMessage().contains("FAILED_TO_RELEASE"));
+    assertTrue(exception.getMessage().contains(StorageBasedLockProvider.CAUSE_INTERRUPTED_DURING_THROTTLE_BACKOFF),
+        exception.getMessage());
+    // The interrupt flag must be re-set so callers up the stack still observe it. Clear it here
+    // so the flag does not leak into subsequent tests on this thread.
+    assertTrue(Thread.interrupted());
+    // Only the initial attempt ran; the interruption aborted the retry loop.
+    verify(mockLockService, times(1)).tryUpsertLockFile(any(), eq(Option.of(realLockFile)));
+    when(mockHeartbeatManager.hasActiveHeartbeat()).thenReturn(false);
+  }
+
   @Test
   void testCloseFailsToStopHeartbeat() {
     when(mockLockService.readCurrentLockFile()).thenReturn(Pair.of(LockGetResult.NOT_EXISTS, Option.empty()));