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 9f8376e3723d fix(hive-sync): keep HMS lock heartbeat alive and release
its thread pool on close (#19334)
9f8376e3723d is described below
commit 9f8376e3723d6e51f5e1a27f52f396b6181a6f6f
Author: Vova Kolmakov <[email protected]>
AuthorDate: Wed Jul 22 17:03:41 2026 +0700
fix(hive-sync): keep HMS lock heartbeat alive and release its thread pool
on close (#19334)
* fix(hive-sync): keep HMS lock heartbeat alive and release its thread pool
on close
Fixes three defects in HiveMetastoreBasedLockProvider and its Heartbeat
(package transaction/lock):
- Heartbeat.run() no longer rethrows on failure. Under
ScheduledExecutorService.scheduleAtFixedRate a thrown exception permanently
cancels all subsequent executions (observable only through the unread
ScheduledFuture), so one transient HMS/network hiccup silently stopped lock
renewal while the writer still believed it held the lock. It now logs a warning
with the cause and lets the next tick retry; the previous throw also discarded
the cause.
- close() moves executor.shutdown() into a finally block so the 2-thread
scheduled pool is always released even when unlock()/Hive.closeCurrent()
throws, and passes the caught exception to log.error instead of dropping it.
- acquireLock recovery path: when the client times out on Future.get but
the lock is granted server-side and recovered via checkLock, a heartbeat is now
scheduled (via the extracted scheduleHeartbeat()) so the recovered lock is
renewed instead of being left to expire mid-write.
Adds unit tests TestHeartbeat and TestHiveMetastoreBasedLockProviderClose.
* addressed review comments: drop the hardcoded thread count from the
executor shutdown comment
---------
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../hudi/hive/transaction/lock/Heartbeat.java | 10 +-
.../lock/HiveMetastoreBasedLockProvider.java | 28 +++--
.../hudi/hive/transaction/lock/TestHeartbeat.java | 55 ++++++++++
.../TestHiveMetastoreBasedLockProviderClose.java | 113 +++++++++++++++++++++
4 files changed, 195 insertions(+), 11 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 f91b66038044..bf0037faf6df 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
@@ -19,10 +19,10 @@
package org.apache.hudi.hive.transaction.lock;
-import org.apache.hudi.exception.HoodieLockException;
-
+import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+@Slf4j
class Heartbeat implements Runnable {
private final IMetaStoreClient client;
private final long lockId;
@@ -37,7 +37,11 @@ class Heartbeat implements Runnable {
try {
client.heartbeat(0, lockId);
} catch (Exception e) {
- throw new HoodieLockException(String.format("Failed to heartbeat for
lock: %d", lockId));
+ // Do not rethrow. This task is scheduled via
ScheduledExecutorService.scheduleAtFixedRate,
+ // where a thrown exception permanently cancels all subsequent
executions and is only
+ // observable through the (unread) ScheduledFuture. Swallowing a
transient failure here keeps
+ // the lock heartbeated on the next tick instead of silently stopping
renewal altogether.
+ log.warn("Failed to heartbeat for lock: {}", lockId, e);
}
}
}
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 80a95801fbdb..c83267be5319 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
@@ -167,9 +167,12 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
future.cancel(false);
}
Hive.closeCurrent();
- executor.shutdown();
} catch (Exception e) {
-
log.error(generateLogStatement(org.apache.hudi.common.lock.LockState.FAILED_TO_RELEASE,
generateLogSuffixString()));
+
log.error(generateLogStatement(org.apache.hudi.common.lock.LockState.FAILED_TO_RELEASE,
generateLogSuffixString()), e);
+ } finally {
+ // Always release the heartbeat thread pool, even if unlock/closeCurrent
above threw,
+ // otherwise its scheduled threads leak for the lifetime of the JVM.
+ executor.shutdown();
}
}
@@ -192,17 +195,15 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
final LockRequest lockRequestFinal = lockRequest;
this.lock = executor.submit(() -> hiveClient.lock(lockRequestFinal))
.get(time, unit);
-
- // refresh lock in case that certain commit takes a long time.
- Heartbeat heartbeat = new Heartbeat(hiveClient, lock.getLockid());
- long heartbeatIntervalMs = lockConfiguration.getConfig()
- .getLong(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
DEFAULT_LOCK_HEARTBEAT_INTERVAL_MS);
- future = executor.scheduleAtFixedRate(heartbeat, heartbeatIntervalMs /
2, heartbeatIntervalMs, TimeUnit.MILLISECONDS);
+ scheduleHeartbeat();
} catch (InterruptedException | TimeoutException e) {
if (this.lock == null || this.lock.getState() != LockState.ACQUIRED) {
LockResponse lockResponse =
this.hiveClient.checkLock(lockRequest.getTxnid());
if (lockResponse.getState() == LockState.ACQUIRED) {
this.lock = lockResponse;
+ // The lock was granted server-side even though the client timed out
waiting on the
+ // future; it still needs a heartbeat, otherwise a long-running
commit lets HMS expire it.
+ scheduleHeartbeat();
} else {
throw e;
}
@@ -219,6 +220,17 @@ public class HiveMetastoreBasedLockProvider implements
LockProvider<LockResponse
}
}
+ /**
+ * 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.
+ */
+ private void scheduleHeartbeat() {
+ Heartbeat heartbeat = new Heartbeat(hiveClient, lock.getLockid());
+ long heartbeatIntervalMs = lockConfiguration.getConfig()
+ .getLong(LOCK_HEARTBEAT_INTERVAL_MS_KEY,
DEFAULT_LOCK_HEARTBEAT_INTERVAL_MS);
+ future = executor.scheduleAtFixedRate(heartbeat, heartbeatIntervalMs / 2,
heartbeatIntervalMs, TimeUnit.MILLISECONDS);
+ }
+
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/TestHeartbeat.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
new file mode 100644
index 000000000000..ca2386f7a28f
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHeartbeat.java
@@ -0,0 +1,55 @@
+/*
+ * 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.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+class TestHeartbeat {
+
+ @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);
+
+ // 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);
+ }
+
+ @Test
+ void runHeartbeatsTheLockOnSuccess() throws TException {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+
+ new Heartbeat(client, 99L).run();
+
+ verify(client, times(1)).heartbeat(0L, 99L);
+ }
+}
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
new file mode 100644
index 000000000000..817c1407f3b9
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderClose.java
@@ -0,0 +1,113 @@
+/*
+ * 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.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;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+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);
+ }
+
+ @Test
+ void closeShutsDownExecutorEvenWhenUnlockThrows() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(42L));
+ doThrow(new TException("boom")).when(client).unlock(anyLong());
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ // 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");
+ }
+
+ @Test
+ void closeShutsDownExecutorOnNormalPath() throws Exception {
+ IMetaStoreClient client = mock(IMetaStoreClient.class);
+ when(client.lock(any())).thenReturn(acquiredLock(1L));
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(lockConfiguration, client);
+ assertTrue(provider.acquireLock(1000L, TimeUnit.MILLISECONDS,
lockComponent));
+
+ 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);
+ }
+}
