voonhous commented on code in PR #19370:
URL: https://github.com/apache/hudi/pull/19370#discussion_r3644453506
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -225,12 +234,33 @@ private void acquireLockInternal(long time, TimeUnit
unit, LockComponent lockCom
* takes a long time. Must be called only after {@link #lock} has been set.
*/
private void scheduleHeartbeat() {
- Heartbeat heartbeat = new Heartbeat(hiveClient, lock.getLockid());
+ Heartbeat heartbeat = new Heartbeat(hiveClient, lock.getLockid(),
this::onLockLost);
Review Comment:
This also gets scheduled for a lock that came back `WAITING` -- the
`finally` at line 224 is about to release it, but the heartbeat is created
first. Before this PR that was harmless; now, if such a tick ever fires before
the cancel, it gets `NoSuchLockException` for the just-released lock and
latches `lockLostRemotely` for a lock we never held, so a later unrelated
`unlock()` throws. Narrow window (first tick is interval/2 away), but there is
no reason to heartbeat a lock we are about to give back.
Call `scheduleHeartbeat()` only when `lock.getState() ==
LockState.ACQUIRED`, and ideally add a WAITING-lock test whose heartbeat throws
`NoSuchLockException`, asserting the later `unlock()` stays silent.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -187,6 +195,7 @@ public boolean acquireLock(long time, TimeUnit unit, final
LockComponent compone
private void acquireLockInternal(long time, TimeUnit unit, LockComponent
lockComponent)
throws InterruptedException, ExecutionException, TimeoutException,
TException {
LockRequest lockRequest = null;
+ lockLostRemotely = false;
Review Comment:
Not this PR's fault, but worth folding into point 1 of #19345 while you are
in here: the recovery branch below has never had a test, and
`checkLock(lockRequest.getTxnid())` at line 210 passes a txn id (never set on
the request, so 0) where `IMetaStoreClient.checkLock(long lockid)` expects a
lock id -- the hive javadoc says it throws `NoSuchLockException` "if the
requested lockid does not exist". So the recovery path most likely throws
instead of recovering, and it now also interacts with this reset and the
heartbeat scheduled at line 215. No action needed in this PR; suggest copying
this onto #19345 so the fix there covers it.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -132,12 +136,16 @@ public void unlock() {
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())
Review Comment:
Design note, not a blocker: this makes `unlock()` throw on the `lock ==
null` path, where providers are otherwise silent -- e.g.
`StorageBasedLockProvider.unlock()` just returns when it does not believe it
holds the lock. For a genuinely lost lock the outcome is the same kind of
failure as master (the stale unlock RPC also ends in `HoodieLockException`), so
the behavior is fine; but since `LockManager.unlock()` skips its metrics and
`close()` when the provider throws, please state the throw explicitly in the
`unlock()` javadoc so the next reader knows it is deliberate.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -159,13 +167,13 @@ public void acquireLock(long time, TimeUnit unit) throws
InterruptedException, E
@Override
public void close() {
try {
- if (lock != null) {
- hiveClient.unlock(lock.getLockid());
+ // Snapshot the lock: the heartbeat thread clears it as soon as the
metastore reports it gone.
+ LockResponse lockResponseLocal = lock;
Review Comment:
The double-read this snapshot fixes still exists in two spots: `tryLock()`
at line 130 and the test-facing `acquireLock` overload at line 192, both
`return this.lock != null && this.lock.getState() == ...`. Not a live NPE in
practice (the first tick that could null `lock` is interval/2 away), but now
that the heartbeat thread clears `lock`, snapshot those two returns the same
way for consistency. Optional.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.hudi.exception.HoodieLockException;
+
+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.hadoop.hive.metastore.api.NoSuchLockException;
+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.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+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;
+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 {
+
+ private static final String DB = "testdb";
+ private static final String TABLE = "testtable";
+ 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;
+
+ 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);
+ // Keep the ticks short so the scheduled heartbeat fires within the test.
+ props.setProperty(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
String.valueOf(HEARTBEAT_INTERVAL_MS));
+ lockConfiguration = new LockConfiguration(props);
+ lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
+ lockComponent.setTablename(TABLE);
+ }
+
+ @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();
Review Comment:
Two cheap coverage extensions, both optional:
- after a loss, `close()` runs via this `finally` but nothing is asserted
about it -- worth checking it does not send a second `unlock` RPC and still
shuts the executor down (mirroring `TestHiveMetastoreBasedLockProviderClose`)
- all five tests use the test-only `acquireLock(time, unit, component)`
overload; one variant driving `tryLock(...)` would cover the entry point
`LockManager` actually uses with the new lost-lock state
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.hudi.exception.HoodieLockException;
+
+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.hadoop.hive.metastore.api.NoSuchLockException;
+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.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+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;
+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 {
+
+ private static final String DB = "testdb";
+ private static final String TABLE = "testtable";
+ 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;
+
+ 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);
+ // Keep the ticks short so the scheduled heartbeat fires within the test.
+ props.setProperty(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
String.valueOf(HEARTBEAT_INTERVAL_MS));
+ lockConfiguration = new LockConfiguration(props);
+ lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
+ lockComponent.setTablename(TABLE);
+ }
+
+ @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 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));
+
+ // Releasing must not report the loss that belonged to the previous lock.
+ 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();
+ }
+ }
+
+ private static ScheduledFuture<?>
heartbeatFutureOf(HiveMetastoreBasedLockProvider provider) throws Exception {
Review Comment:
Nit, feel free to ignore: this reflection reader, the `DB`/`TABLE`
constants, `setUp()`, and the `LockResponse` factories duplicate
`TestHiveMetastoreBasedLockProviderClose` (~25-30 lines). A small
package-private base class (constants + fixture + lock-response factory + one
generic `readField(provider, name)` helper) would serve both files.
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.hudi.exception.HoodieLockException;
+
+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.hadoop.hive.metastore.api.NoSuchLockException;
+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.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+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;
+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 {
+
+ private static final String DB = "testdb";
+ private static final String TABLE = "testtable";
+ 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;
+
+ 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);
+ // Keep the ticks short so the scheduled heartbeat fires within the test.
+ props.setProperty(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
String.valueOf(HEARTBEAT_INTERVAL_MS));
+ lockConfiguration = new LockConfiguration(props);
+ lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
+ lockComponent.setTablename(TABLE);
+ }
+
+ @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 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));
Review Comment:
As-is these assertions would pass even without the snapshot change in
`acquireLockInternal`'s finally -- nothing checks the WAITING lock was actually
released. One extra line makes the test discriminate it:
```suggestion
assertFalse(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
verify(client).unlock(OTHER_LOCK_ID);
```
##########
hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderLockLoss.java:
##########
@@ -0,0 +1,244 @@
+/*
+ * 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.hudi.exception.HoodieLockException;
+
+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.hadoop.hive.metastore.api.NoSuchLockException;
+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.CountDownLatch;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+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;
+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 {
+
+ private static final String DB = "testdb";
+ private static final String TABLE = "testtable";
+ 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;
+
+ 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);
+ // Keep the ticks short so the scheduled heartbeat fires within the test.
+ props.setProperty(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
String.valueOf(HEARTBEAT_INTERVAL_MS));
+ lockConfiguration = new LockConfiguration(props);
+ lockComponent = new LockComponent(LockType.EXCLUSIVE, LockLevel.TABLE, DB);
+ lockComponent.setTablename(TABLE);
+ }
+
+ @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");
Review Comment:
Tiny race: `onLockLost` nulls `lock` before it cancels the future, and
`awaitUntil` above only waits for `lock == null`, so this assert can in
principle run between the two. The window is nanoseconds, but making it
deterministic is free: either cancel before clearing `lock` in `onLockLost`
(keeping the `lockLostRemotely` write first), or await the cancellation here
the same way as the lock drop above. Optional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
