This is an automated email from the ASF dual-hosted git repository.
voonhous 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 452b237501df fix(lock): name the cause on FAILED_TO_RELEASE in
StorageBasedLockProvider (#19574)
452b237501df is described below
commit 452b237501df5187f486856bc55369f83f1b6c64
Author: Praveen Gajulapalli <[email protected]>
AuthorDate: Fri Aug 28 09:25:17 2026 +0530
fix(lock): name the cause on FAILED_TO_RELEASE in StorageBasedLockProvider
(#19574)
* 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.
* Include cause in the logs
---
.../transaction/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 974ea01aee7a..fc6d5d848aa7 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 class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
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 class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
// 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 class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
}
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
{} (cause={}) after {} "
+ + "throttle retries; lock left un-expired and will dangle until
its lease elapses.",
+ ownerId, lockFilePath, expireResult, cause, 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 @@ public class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
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 @@ public class StorageBasedLockProvider implements
LockProvider<StorageLockFile> {
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 85c21c129d0f..5a9775589dca 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 @@ class TestStorageBasedLockProvider {
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 @@ class TestStorageBasedLockProvider {
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 @@ class TestStorageBasedLockProvider {
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 @@ class TestStorageBasedLockProvider {
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()));
