fix(lock): name the cause on FAILED_TO_RELEASE in StorageBasedLockProvider

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 {}, {}";
 

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()));