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