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 48fe2c6bd0a4 fix(hive-sync): stop the HMS lock heartbeat once the
metastore drops the lock (#19370)
48fe2c6bd0a4 is described below
commit 48fe2c6bd0a44fa1ef5693aa7f7460c32f0cfd61
Author: Vova Kolmakov <[email protected]>
AuthorDate: Fri Jul 24 19:50:07 2026 +0700
fix(hive-sync): stop the HMS lock heartbeat once the metastore drops the
lock (#19370)
* fix(hive-sync): stop the HMS lock heartbeat once the metastore drops the
lock
* addressed review comments: scope the lock-loss callback to its lock id
* addressed review comments: document the unlock throw, snapshot lock reads
and extend lock-loss coverage
* addressed review comments: skip the heartbeat for a lock that was never
granted
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../hudi/hive/transaction/lock/Heartbeat.java | 22 +-
.../lock/HiveMetastoreBasedLockProvider.java | 117 +++++--
.../HiveMetastoreBasedLockProviderTestBase.java | 95 ++++++
.../hudi/hive/transaction/lock/TestHeartbeat.java | 73 ++++-
.../TestHiveMetastoreBasedLockProviderClose.java | 50 +--
...TestHiveMetastoreBasedLockProviderLockLoss.java | 362 +++++++++++++++++++++
6 files changed, 652 insertions(+), 67 deletions(-)
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/Heartbeat.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/Heartbeat.java
index bf0037faf6df..0766419d677a 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/Heartbeat.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/Heartbeat.java
@@ -21,21 +21,41 @@ package org.apache.hudi.hive.transaction.lock;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
+
+import java.util.function.Consumer;
@Slf4j
class Heartbeat implements Runnable {
private final IMetaStoreClient client;
private final long lockId;
+ private final Consumer<Exception> onLockLost;
+ // Latches the terminal failure so that a tick already queued when the
schedule was cancelled
+ // does not issue another doomed heartbeat or report the loss a second time.
+ private volatile boolean lockLost = false;
- Heartbeat(IMetaStoreClient client, long lockId) {
+ Heartbeat(IMetaStoreClient client, long lockId, Consumer<Exception>
onLockLost) {
this.client = client;
this.lockId = lockId;
+ this.onLockLost = onLockLost;
}
@Override
public void run() {
+ if (lockLost) {
+ return;
+ }
try {
client.heartbeat(0, lockId);
+ } catch (NoSuchLockException | NoSuchTxnException | TxnAbortedException e)
{
+ // Terminal. The metastore has already expired or aborted this lock, so
no later tick can
+ // renew it. Retrying would only log once per interval while the writer
keeps believing it
+ // holds exclusivity, so report the loss to the owner, which stops the
schedule and drops
+ // the lock.
+ lockLost = true;
+ onLockLost.accept(e);
} catch (Exception e) {
// Do not rethrow. This task is scheduled via
ScheduledExecutorService.scheduleAtFixedRate,
// where a thrown exception permanently cancels all subsequent
executions and is only
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java
index c83267be5319..91d4062ecc3b 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java
@@ -87,7 +87,11 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
@Getter
private volatile LockResponse lock = null;
protected LockConfiguration lockConfiguration;
- private transient ScheduledFuture<?> future = null;
+ // Assigned by the acquiring thread, read and cancelled by the heartbeat
thread.
+ private transient volatile ScheduledFuture<?> future = null;
+ // Set when the metastore reports the lock as expired or aborted, so that a
later unlock() can
+ // tell the caller its exclusivity was lost instead of silently doing
nothing.
+ private volatile boolean lockLostRemotely = false;
private final transient ScheduledExecutorService executor =
Executors.newScheduledThreadPool(2);
public HiveMetastoreBasedLockProvider(final LockConfiguration
lockConfiguration, final StorageConfiguration<?> conf) {
@@ -123,21 +127,35 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
} catch (ExecutionException | InterruptedException | TimeoutException |
TException e) {
throw new HoodieLockException(generateLogStatement(FAILED_TO_ACQUIRE,
generateLogSuffixString()), e);
}
- return this.lock != null && this.lock.getState() == LockState.ACQUIRED;
+ return isLockAcquired();
}
+ /**
+ * Releases the lock held at the metastore, if any.
+ *
+ * <p>Unlike most providers, which stay silent when they do not believe they
hold a lock, this one
+ * deliberately fails when the metastore had already expired or aborted the
lock: the writer went
+ * on committing without exclusivity and has to hear about it. Note that
{@code LockManager.unlock()}
+ * skips its metrics and its {@code close()} when the provider throws.
+ *
+ * @throws HoodieLockException if the metastore took the lock away, or if
releasing it fails.
+ */
@Override
public void unlock() {
try {
log.info(generateLogStatement(RELEASING, generateLogSuffixString()));
LockResponse lockResponseLocal = lock;
if (lockResponseLocal == null) {
+ if (lockLostRemotely) {
+ // The heartbeat already dropped the lock. Unlocking it would fail
with a bare
+ // NoSuchLockException anyway, so fail with the actual reason
instead.
+ throw new
HoodieLockException(generateLogStatement(FAILED_TO_RELEASE,
generateLogSuffixString())
+ + ", the metastore had already expired or aborted it");
+ }
return;
}
lock = null;
- if (future != null) {
- future.cancel(false);
- }
+ cancelHeartbeat();
hiveClient.unlock(lockResponseLocal.getLockid());
log.info(generateLogStatement(RELEASED, generateLogSuffixString()));
} catch (TException e) {
@@ -159,12 +177,14 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
@Override
public void close() {
try {
- if (lock != null) {
- hiveClient.unlock(lock.getLockid());
- lock = null;
- }
- if (future != null) {
- future.cancel(false);
+ // Snapshot the lock, then stop claiming it before releasing it, exactly
as unlock() does:
+ // the release itself makes an in-flight heartbeat fail, and that must
not be mistaken for
+ // the metastore taking the lock away.
+ LockResponse lockResponseLocal = lock;
+ lock = null;
+ cancelHeartbeat();
+ if (lockResponseLocal != null) {
+ hiveClient.unlock(lockResponseLocal.getLockid());
}
Hive.closeCurrent();
} catch (Exception e) {
@@ -181,12 +201,22 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
throws InterruptedException, ExecutionException, TimeoutException,
TException {
ValidationUtils.checkArgument(this.lock == null, ALREADY_ACQUIRED.name());
acquireLockInternal(time, unit, component);
- return this.lock != null && this.lock.getState() == LockState.ACQUIRED;
+ return isLockAcquired();
+ }
+
+ /**
+ * Whether the lock is held right now. Reads {@link #lock} once: the
heartbeat thread clears it
+ * as soon as the metastore reports the lock as gone, so re-reading the
field can mix two states.
+ */
+ private boolean isLockAcquired() {
+ LockResponse lockResponseLocal = this.lock;
+ return lockResponseLocal != null && lockResponseLocal.getState() ==
LockState.ACQUIRED;
}
private void acquireLockInternal(long time, TimeUnit unit, LockComponent
lockComponent)
throws InterruptedException, ExecutionException, TimeoutException,
TException {
LockRequest lockRequest = null;
+ lockLostRemotely = false;
try {
// TODO : FIX:Using the parameterized constructor throws MethodNotFound
final LockRequestBuilder builder = new LockRequestBuilder();
@@ -210,27 +240,76 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
}
} finally {
// it is better to release WAITING lock, otherwise hive lock will hang
forever
- if (this.lock != null && this.lock.getState() != LockState.ACQUIRED) {
- hiveClient.unlock(this.lock.getLockid());
+ // Snapshot the lock: the heartbeat thread clears it as soon as the
metastore reports it gone.
+ LockResponse lockResponseLocal = this.lock;
+ if (lockResponseLocal != null && lockResponseLocal.getState() !=
LockState.ACQUIRED) {
+ hiveClient.unlock(lockResponseLocal.getLockid());
lock = null;
- if (future != null) {
- future.cancel(false);
- }
+ cancelHeartbeat();
}
}
}
/**
* Schedules a periodic {@link Heartbeat} to refresh the currently held lock
in case a commit
- * takes a long time. Must be called only after {@link #lock} has been set.
+ * takes a long time. Does nothing unless {@link #lock} is held right now,
so that callers may
+ * invoke it without checking the state of the response they just got.
*/
private void scheduleHeartbeat() {
- Heartbeat heartbeat = new Heartbeat(hiveClient, lock.getLockid());
+ LockResponse lockResponseLocal = lock;
+ if (lockResponseLocal == null) {
+ // Released while the acquisition was still completing, so there is
nothing left to renew.
+ return;
+ }
+ if (lockResponseLocal.getState() != LockState.ACQUIRED) {
+ // The metastore only queued the request. The caller releases such a
lock right away, and
+ // renewing one that was never granted can only latch a loss that never
happened.
+ return;
+ }
+ // Bind the id into the task and its callback: cancelling a schedule does
not stop a tick that
+ // is already inside the heartbeat RPC, so a tick can outlive the lock it
was scheduled for.
+ long lockId = lockResponseLocal.getLockid();
+ Heartbeat heartbeat = new Heartbeat(hiveClient, lockId, cause ->
onLockLost(lockId, cause));
long heartbeatIntervalMs = lockConfiguration.getConfig()
.getLong(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
DEFAULT_LOCK_HEARTBEAT_INTERVAL_MS);
future = executor.scheduleAtFixedRate(heartbeat, heartbeatIntervalMs / 2,
heartbeatIntervalMs, TimeUnit.MILLISECONDS);
}
+ /**
+ * Invoked from the heartbeat thread once the metastore reports the lock as
expired or aborted.
+ * The lock cannot be renewed anymore, so stop heartbeating it and stop
claiming it is held.
+ *
+ * @param lockId the lock this heartbeat was scheduled for, which is not
necessarily the one held
+ * now: releasing a lock is itself a reason for an in-flight
heartbeat to fail.
+ */
+ private void onLockLost(long lockId, Exception cause) {
+ LockResponse lockResponseLocal = lock;
+ if (lockResponseLocal == null || lockResponseLocal.getLockid() != lockId) {
+ // We released this lock ourselves while the tick was in flight, which
is why the metastore
+ // no longer knows about it. Nothing was lost, and any lock held now is
a different one that
+ // keeps its own heartbeat.
+ log.debug("Ignoring a heartbeat failure for the already released lock
{}", lockId, cause);
+ return;
+ }
+ // Order matters: the flag is set first, so an unlock() racing with this
can never find the
+ // lock gone without a reason for it; and the schedule is cancelled before
the lock is dropped,
+ // so whoever observes the drop is guaranteed to see the renewal already
stopped.
+ lockLostRemotely = true;
+ cancelHeartbeat();
+ lock = null;
+ log.error("The metastore expired or aborted the lock at{}, heartbeat
stopped and exclusivity is lost",
+ generateLogSuffixString(), cause);
+ }
+
+ private void cancelHeartbeat() {
+ ScheduledFuture<?> futureLocal = future;
+ if (futureLocal != null) {
+ // Never interrupt: this can run on the heartbeat thread itself, and the
current tick is
+ // harmless. Cancelling only prevents further executions.
+ futureLocal.cancel(false);
+ }
+ }
+
private void checkRequiredProps(final LockConfiguration lockConfiguration) {
ValidationUtils.checkArgument(lockConfiguration.getConfig().getString(HIVE_DATABASE_NAME_PROP_KEY)
!= null);
ValidationUtils.checkArgument(lockConfiguration.getConfig().getString(HIVE_TABLE_NAME_PROP_KEY)
!= null);
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProviderTestBase.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProviderTestBase.java
new file mode 100644
index 000000000000..0e329e3b2f45
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProviderTestBase.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.hive.transaction.lock;
+
+import org.apache.hudi.common.config.LockConfiguration;
+import org.apache.hudi.common.config.TypedProperties;
+
+import org.apache.hadoop.hive.metastore.api.LockComponent;
+import org.apache.hadoop.hive.metastore.api.LockLevel;
+import org.apache.hadoop.hive.metastore.api.LockResponse;
+import org.apache.hadoop.hive.metastore.api.LockState;
+import org.apache.hadoop.hive.metastore.api.LockType;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.lang.reflect.Field;
+
+import static
org.apache.hudi.common.config.LockConfiguration.DEFAULT_LOCK_HEARTBEAT_INTERVAL_MS;
+import static
org.apache.hudi.common.config.LockConfiguration.HIVE_DATABASE_NAME_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.HIVE_TABLE_NAME_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.LOCK_HEARTBEAT_INTERVAL_MS_KEY;
+
+/**
+ * Shared fixture for the {@link HiveMetastoreBasedLockProvider} unit tests
that drive the provider
+ * against a mocked {@code IMetaStoreClient}, without a live metastore or
ZooKeeper.
+ */
+abstract class HiveMetastoreBasedLockProviderTestBase {
+
+ protected static final String DB = "testdb";
+ protected static final String TABLE = "testtable";
+
+ protected LockConfiguration lockConfiguration;
+ protected LockComponent lockComponent;
+
+ @BeforeEach
+ void setUpLockFixture() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(HIVE_DATABASE_NAME_PROP_KEY, DB);
+ props.setProperty(HIVE_TABLE_NAME_PROP_KEY, TABLE);
+ props.setProperty(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
String.valueOf(heartbeatIntervalMs()));
+ lockConfiguration = new LockConfiguration(props);
+ lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
+ lockComponent.setTablename(TABLE);
+ }
+
+ /**
+ * The heartbeat interval the provider is configured with, overridden by
tests that need the
+ * scheduled heartbeat to actually fire while the test runs.
+ */
+ protected long heartbeatIntervalMs() {
+ return DEFAULT_LOCK_HEARTBEAT_INTERVAL_MS;
+ }
+
+ protected static LockResponse acquiredLock(long lockId) {
+ return lockResponse(lockId, LockState.ACQUIRED);
+ }
+
+ protected static LockResponse waitingLock(long lockId) {
+ return lockResponse(lockId, LockState.WAITING);
+ }
+
+ private static LockResponse lockResponse(long lockId, LockState state) {
+ LockResponse response = new LockResponse();
+ response.setLockid(lockId);
+ response.setState(state);
+ return response;
+ }
+
+ /**
+ * Reads a private field of the provider, for the state it does not expose:
the heartbeat
+ * schedule and the thread pool running it.
+ */
+ @SuppressWarnings("unchecked")
+ protected static <T> T readField(HiveMetastoreBasedLockProvider provider,
String name) throws Exception {
+ Field field = HiveMetastoreBasedLockProvider.class.getDeclaredField(name);
+ field.setAccessible(true);
+ return (T) field.get(provider);
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
index ca2386f7a28f..2dbdf827de87 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
@@ -20,10 +20,22 @@
package org.apache.hudi.hive.transaction.lock;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.hadoop.hive.metastore.api.NoSuchTxnException;
+import org.apache.hadoop.hive.metastore.api.TxnAbortedException;
import org.apache.thrift.TException;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -32,24 +44,81 @@ import static org.mockito.Mockito.verify;
class TestHeartbeat {
+ private final List<Exception> lockLostCauses = new ArrayList<>();
+
+ /**
+ * Failures that mean the metastore has already expired or aborted the lock.
They are declared by
+ * {@code IMetaStoreClient.heartbeat(long, long)} and cannot be recovered
from by retrying.
+ */
+ private static Stream<Exception> terminalFailures() {
+ return Stream.of(
+ new NoSuchLockException("lock does not exist"),
+ new NoSuchTxnException("txn does not exist"),
+ new TxnAbortedException("txn was aborted"));
+ }
+
@Test
void runDoesNotRethrowWhenHeartbeatFails() throws TException {
IMetaStoreClient client = mock(IMetaStoreClient.class);
doThrow(new TException("transient
failure")).when(client).heartbeat(anyLong(), anyLong());
- Heartbeat heartbeat = new Heartbeat(client, 7L);
+ Heartbeat heartbeat = new Heartbeat(client, 7L, lockLostCauses::add);
// Rethrowing here would cancel every subsequent execution of a
scheduleAtFixedRate task,
// silently stopping lock renewal. The fix must swallow the failure so the
next tick retries.
assertDoesNotThrow(heartbeat::run);
+ assertTrue(lockLostCauses.isEmpty(), "a transient failure must not be
reported as a lost lock");
+ }
+
+ @Test
+ void runKeepsRetryingAfterTransientFailure() throws TException {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ doThrow(new TException("transient
failure")).when(client).heartbeat(anyLong(), anyLong());
+
+ Heartbeat heartbeat = new Heartbeat(client, 7L, lockLostCauses::add);
+ heartbeat.run();
+ heartbeat.run();
+
+ verify(client, times(2)).heartbeat(0L, 7L);
+ assertTrue(lockLostCauses.isEmpty());
}
@Test
void runHeartbeatsTheLockOnSuccess() throws TException {
IMetaStoreClient client = mock(IMetaStoreClient.class);
- new Heartbeat(client, 99L).run();
+ new Heartbeat(client, 99L, lockLostCauses::add).run();
verify(client, times(1)).heartbeat(0L, 99L);
+ assertTrue(lockLostCauses.isEmpty());
+ }
+
+ @ParameterizedTest
+ @MethodSource("terminalFailures")
+ void runReportsLockLossOnTerminalFailure(Exception terminalFailure) throws
TException {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ doThrow(terminalFailure).when(client).heartbeat(anyLong(), anyLong());
+
+ Heartbeat heartbeat = new Heartbeat(client, 11L, lockLostCauses::add);
+
+ assertDoesNotThrow(heartbeat::run);
+ assertEquals(1, lockLostCauses.size());
+ assertSame(terminalFailure, lockLostCauses.get(0));
+ }
+
+ @ParameterizedTest
+ @MethodSource("terminalFailures")
+ void runStopsHeartbeatingAfterTerminalFailure(Exception terminalFailure)
throws TException {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ doThrow(terminalFailure).when(client).heartbeat(anyLong(), anyLong());
+
+ Heartbeat heartbeat = new Heartbeat(client, 11L, lockLostCauses::add);
+ heartbeat.run();
+ // A tick already queued when the schedule was cancelled must not renew a
lock that is gone,
+ // nor report the loss a second time.
+ heartbeat.run();
+
+ verify(client, times(1)).heartbeat(0L, 11L);
+ assertEquals(1, lockLostCauses.size());
}
}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderClose.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderClose.java
index 817c1407f3b9..3d3a8c26c4e8 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderClose.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderClose.java
@@ -19,25 +19,13 @@
package org.apache.hudi.hive.transaction.lock;
-import org.apache.hudi.common.config.LockConfiguration;
-import org.apache.hudi.common.config.TypedProperties;
-
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.LockComponent;
-import org.apache.hadoop.hive.metastore.api.LockLevel;
-import org.apache.hadoop.hive.metastore.api.LockResponse;
-import org.apache.hadoop.hive.metastore.api.LockState;
-import org.apache.hadoop.hive.metastore.api.LockType;
import org.apache.thrift.TException;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.lang.reflect.Field;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import static
org.apache.hudi.common.config.LockConfiguration.HIVE_DATABASE_NAME_PROP_KEY;
-import static
org.apache.hudi.common.config.LockConfiguration.HIVE_TABLE_NAME_PROP_KEY;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -51,23 +39,7 @@ import static org.mockito.Mockito.when;
* Unit tests for {@link HiveMetastoreBasedLockProvider#close()} that exercise
the thread-pool
* shutdown path with a mocked {@link IMetaStoreClient}, without a live
metastore or ZooKeeper.
*/
-class TestHiveMetastoreBasedLockProviderClose {
-
- private static final String DB = "testdb";
- private static final String TABLE = "testtable";
-
- private LockConfiguration lockConfiguration;
- private LockComponent lockComponent;
-
- @BeforeEach
- void setUp() {
- TypedProperties props = new TypedProperties();
- props.setProperty(HIVE_DATABASE_NAME_PROP_KEY, DB);
- props.setProperty(HIVE_TABLE_NAME_PROP_KEY, TABLE);
- lockConfiguration = new LockConfiguration(props);
- lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
- lockComponent.setTablename(TABLE);
- }
+class TestHiveMetastoreBasedLockProviderClose extends
HiveMetastoreBasedLockProviderTestBase {
@Test
void closeShutsDownExecutorEvenWhenUnlockThrows() throws Exception {
@@ -80,8 +52,8 @@ class TestHiveMetastoreBasedLockProviderClose {
// A failing unlock() must not prevent the heartbeat thread pool from
being shut down.
assertDoesNotThrow(provider::close);
- assertTrue(executorOf(provider).isShutdown(),
- "executor must be shut down even when unlock() throws");
+ ScheduledExecutorService executor = readField(provider, "executor");
+ assertTrue(executor.isShutdown(), "executor must be shut down even when
unlock() throws");
}
@Test
@@ -95,19 +67,7 @@ class TestHiveMetastoreBasedLockProviderClose {
provider.close();
verify(client).unlock(1L);
- assertTrue(executorOf(provider).isShutdown());
- }
-
- private static LockResponse acquiredLock(long lockId) {
- LockResponse response = new LockResponse();
- response.setLockid(lockId);
- response.setState(LockState.ACQUIRED);
- return response;
- }
-
- private static ScheduledExecutorService
executorOf(HiveMetastoreBasedLockProvider provider) throws Exception {
- Field field =
HiveMetastoreBasedLockProvider.class.getDeclaredField("executor");
- field.setAccessible(true);
- return (ScheduledExecutorService) field.get(provider);
+ ScheduledExecutorService executor = readField(provider, "executor");
+ assertTrue(executor.isShutdown());
}
}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java
new file mode 100644
index 000000000000..b485f4b9cc39
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java
@@ -0,0 +1,362 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.hive.transaction.lock;
+
+import org.apache.hudi.exception.HoodieLockException;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchLockException;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for how {@link HiveMetastoreBasedLockProvider} reacts to the
metastore reporting its
+ * lock as gone, with a mocked {@link IMetaStoreClient} and no live metastore
or ZooKeeper.
+ */
+class TestHiveMetastoreBasedLockProviderLockLoss extends
HiveMetastoreBasedLockProviderTestBase {
+
+ private static final long LOCK_ID = 42L;
+ private static final long OTHER_LOCK_ID = 43L;
+ private static final long HEARTBEAT_INTERVAL_MS = 100L;
+ private static final long AWAIT_TIMEOUT_MS = 30_000L;
+
+ @Override
+ protected long heartbeatIntervalMs() {
+ // Keep the ticks short so the scheduled heartbeat fires within the test.
+ return HEARTBEAT_INTERVAL_MS;
+ }
+
+ @Test
+ void terminalHeartbeatFailureStopsRenewalAndDropsTheLock() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID));
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " does not exist"))
+ .when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ // The metastore has expired the lock, so the provider must stop
claiming to hold it.
+ awaitUntil(() -> provider.getLock() == null, "the lost lock must be
dropped by the heartbeat");
+
+ // The heartbeat task latches the failure on its own, so the count below
would stay at one
+ // even if the schedule were left running. Assert the cancellation
itself as well.
+ assertTrue(heartbeatFutureOf(provider).isCancelled(), "the heartbeat
schedule must be cancelled");
+
+ // Give the scheduler several more intervals: no further heartbeat may
be attempted, since
+ // both the schedule and the heartbeat task itself are stopped after a
terminal failure.
+ Thread.sleep(HEARTBEAT_INTERVAL_MS * 5);
+ verify(client, times(1)).heartbeat(0L, LOCK_ID);
+
+ // The writer must learn that it no longer holds exclusivity, and the
provider must not send
+ // a doomed unlock for a lock the metastore has already dropped.
+ assertThrows(HoodieLockException.class, provider::unlock);
+ verify(client, never()).unlock(anyLong());
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void lockLostAfterTryLockIsReportedOnUnlock() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID));
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " does not exist"))
+ .when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ // Same loss, but driven through the entry point the LockManager
actually calls rather than
+ // the test-only acquireLock overload.
+ assertTrue(provider.tryLock(1000L, TimeUnit.MILLISECONDS));
+ awaitUntil(() -> provider.getLock() == null, "the lost lock must be
dropped by the heartbeat");
+
+ assertThrows(HoodieLockException.class, provider::unlock);
+ verify(client, never()).unlock(anyLong());
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void closeAfterALostLockSendsNoUnlockAndShutsDownTheExecutor() throws
Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID));
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " does not exist"))
+ .when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ awaitUntil(() -> provider.getLock() == null, "the lost lock must be
dropped by the heartbeat");
+ } finally {
+ provider.close();
+ }
+
+ // There is nothing left to release at the metastore, but the heartbeat
pool must still go away.
+ verify(client, never()).unlock(anyLong());
+ ScheduledExecutorService executor = readField(provider, "executor");
+ assertTrue(executor.isShutdown(), "the heartbeat pool must be shut down
after a lost lock");
+ }
+
+ @Test
+ void transientHeartbeatFailureKeepsRenewingTheLock() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID));
+ CountDownLatch heartbeats = new CountDownLatch(2);
+ doAnswer(invocation -> {
+ heartbeats.countDown();
+ throw new TException("transient failure");
+ }).when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ assertTrue(heartbeats.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS),
+ "a transient failure must not stop the heartbeat schedule");
+ assertNotNull(provider.getLock(), "a transient failure must not drop the
lock");
+
+ provider.unlock();
+ verify(client).unlock(LOCK_ID);
+ assertNull(provider.getLock());
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void lockCanBeAcquiredAgainAfterItWasLost() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID),
acquiredLock(OTHER_LOCK_ID));
+ // Only the first lock is expired by the metastore; heartbeating the
second one succeeds.
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " does not exist"))
+ .doNothing()
+ .when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ awaitUntil(() -> provider.getLock() == null, "the lost lock must be
dropped by the heartbeat");
+
+ // Acquiring again must clear the lost-lock state, otherwise the
provider would keep failing
+ // to release locks it holds perfectly well.
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ provider.unlock();
+
+ verify(client).unlock(OTHER_LOCK_ID);
+ assertNull(provider.getLock());
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void lostLockStateDoesNotLeakIntoTheNextAcquire() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID),
waitingLock(OTHER_LOCK_ID));
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " does not exist"))
+ .doNothing()
+ .when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ awaitUntil(() -> provider.getLock() == null, "the lost lock must be
dropped by the heartbeat");
+
+ // The second attempt only got queued, so it leaves no lock behind and
nothing was expired
+ // by the metastore this time round.
+ assertFalse(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ verify(client).unlock(OTHER_LOCK_ID);
+
+ // Releasing must not report the loss that belonged to the previous lock.
+ assertDoesNotThrow(provider::unlock);
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void lockThatWasOnlyQueuedIsNeverHeartbeated() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(waitingLock(LOCK_ID));
+ doThrow(new NoSuchLockException("lock " + LOCK_ID + " was never granted"))
+ .when(client).heartbeat(anyLong(), anyLong());
+ // Releasing the queued lock is the provider's very next step, and it is
an RPC: parking inside
+ // it holds open exactly the window in which a heartbeat scheduled for
that lock would tick.
+ doAnswer(invocation -> {
+ Thread.sleep(HEARTBEAT_INTERVAL_MS * 5);
+ return null;
+ }).when(client).unlock(anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertFalse(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ // A lock that was never granted must never be renewed: a tick failing
for it would be read
+ // as the metastore taking away exclusivity that the writer never had in
the first place.
+ verify(client, never()).heartbeat(anyLong(), anyLong());
+ assertNull(heartbeatFutureOf(provider), "a queued lock must not be given
a heartbeat schedule");
+ assertDoesNotThrow(provider::unlock);
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void unlockStaysSilentWhenNoLockWasEverHeld() {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ // Releasing a lock that was never acquired is still a no-op: only a
lock the metastore took
+ // away is reported as a failure.
+ assertDoesNotThrow(provider::unlock);
+ } finally {
+ provider.close();
+ }
+ }
+
+ @Test
+ void staleHeartbeatFromAReleasedLockDoesNotDropTheNextLock() throws
Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID),
acquiredLock(OTHER_LOCK_ID));
+ CountDownLatch staleTickStarted = new CountDownLatch(1);
+ CountDownLatch releaseStaleTick = new CountDownLatch(1);
+ CountDownLatch newLockHeartbeats = new CountDownLatch(3);
+ doAnswer(invocation -> {
+ long heartbeatedLockId = invocation.getArgument(1);
+ if (heartbeatedLockId == LOCK_ID) {
+ // Park this tick inside the RPC until the lock it renews has been
released and another one
+ // taken, then fail it the way the metastore fails a heartbeat for a
lock it no longer has.
+ staleTickStarted.countDown();
+ releaseStaleTick.await();
+ throw new NoSuchLockException("lock " + LOCK_ID + " does not exist");
+ }
+ newLockHeartbeats.countDown();
+ return null;
+ }).when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ assertTrue(staleTickStarted.await(AWAIT_TIMEOUT_MS,
TimeUnit.MILLISECONDS),
+ "the heartbeat for the first lock must be in flight before it is
released");
+
+ provider.unlock();
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ // Only now does the tick scheduled for the first lock come back,
failing because that lock
+ // was released here. Cancelling its schedule could never have stopped
it.
+ releaseStaleTick.countDown();
+
+ assertTrue(newLockHeartbeats.await(AWAIT_TIMEOUT_MS,
TimeUnit.MILLISECONDS),
+ "the stale tick must not stop the renewal of the lock held now,
which would let the "
+ + "metastore expire a perfectly healthy lock mid-commit");
+ assertNotNull(provider.getLock(), "the stale tick must not drop the lock
held now");
+ assertFalse(heartbeatFutureOf(provider).isCancelled(),
+ "the stale tick must not cancel the schedule of the lock held now");
+
+ assertDoesNotThrow(provider::unlock);
+ verify(client).unlock(OTHER_LOCK_ID);
+ } finally {
+ // Free the parked tick even when an assertion above failed: close()
only shuts the executor
+ // down, which neither interrupts the await nor lets the non-daemon
thread exit, so leaving
+ // it parked would turn a failing test into a hanging JVM.
+ releaseStaleTick.countDown();
+ provider.close();
+ }
+ }
+
+ @Test
+ void releasingALockNormallyIsNotReportedAsLost() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(LOCK_ID));
+ CountDownLatch tickStarted = new CountDownLatch(1);
+ CountDownLatch releaseTick = new CountDownLatch(1);
+ doAnswer(invocation -> {
+ tickStarted.countDown();
+ releaseTick.await();
+ throw new NoSuchLockException("lock " + LOCK_ID + " does not exist");
+ }).when(client).heartbeat(anyLong(), anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ try {
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+ assertTrue(tickStarted.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS),
+ "the heartbeat must be in flight before the lock is released");
+
+ provider.unlock();
+ verify(client).unlock(LOCK_ID);
+ releaseTick.countDown();
+
+ // Let the released tick finish and run whatever it makes of the failure.
+ Thread.sleep(HEARTBEAT_INTERVAL_MS * 5);
+
+ // The heartbeat failed only because the lock was released here, so
releasing again stays the
+ // no-op it has always been instead of reporting a loss that never
happened.
+ assertDoesNotThrow(provider::unlock);
+ } finally {
+ // See the note in
staleHeartbeatFromAReleasedLockDoesNotDropTheNextLock: a parked tick must
+ // never outlive a failed assertion.
+ releaseTick.countDown();
+ provider.close();
+ }
+ }
+
+ private static ScheduledFuture<?>
heartbeatFutureOf(HiveMetastoreBasedLockProvider provider) throws Exception {
+ return readField(provider, "future");
+ }
+
+ private static void awaitUntil(BooleanSupplier condition, String message)
throws InterruptedException {
+ long deadline = System.currentTimeMillis() + AWAIT_TIMEOUT_MS;
+ while (System.currentTimeMillis() < deadline) {
+ if (condition.getAsBoolean()) {
+ return;
+ }
+ Thread.sleep(20L);
+ }
+ fail(message);
+ }
+}
