hudi-agent commented on code in PR #19334:
URL: https://github.com/apache/hudi/pull/19334#discussion_r3624605693
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -192,17 +195,15 @@ private void acquireLockInternal(long time, TimeUnit
unit, LockComponent lockCom
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();
Review Comment:
🤖 On this timeout-recovery path the `hiveClient.lock(...)` task submitted at
line 196 may still be running on a pool thread (the `Future` from
`executor.submit` isn't cancelled after the `TimeoutException`). Scheduling a
recurring heartbeat here means `hiveClient.heartbeat(...)` can now fire
concurrently with that still-blocked `lock()` call. Since `IMetaStoreClient`
(thrift) isn't thread-safe, could this corrupt the client? Would cancelling the
original lock future before scheduling the heartbeat here make this safer?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/transaction/lock/HiveMetastoreBasedLockProvider.java:
##########
@@ -167,9 +167,12 @@ public void close() {
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 the 2 scheduled threads leak for the lifetime of the JVM.
Review Comment:
🤖 nit: the `2` in "the 2 scheduled threads" is a magic number with no clear
origin -- if the executor pool size ever changes, this comment silently lies.
Could you drop the count entirely (e.g. "otherwise scheduled heartbeat threads
leak...") or tie it to whatever constant drives the pool size?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
