HIVE-29237: Addendum: Hive: Refactor commit lock mechanism from HiveTableOperations (#6112)

Note: this isn’t a full port — it brings over some missing changes from the previous port, mainly related to tests.

(cherry picked from commit 333227fbd13821365cec1bdbfcb9314a239bea0f)

Author: InvisibleProgrammer

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

@@ -304,16 +304,6 @@ public ClientPool<IMetaStoreClient, TException> metaClients() {
     return metaClients;
   }
 
-  void doUnlock(HiveLock lock) {
-    if (lock != null) {
-      try {
-        lock.unlock();
-      } catch (Exception e) {
-        LOG.warn("Failed to unlock {}.{}", database, tableName, e);
-      }
-    }
-  }
-
   /**
    * Returns if the hive engine related values should be enabled on the table, or not.
    * <p>

iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java

@@ -281,7 +281,6 @@ public void testLockAcquisitionAfterFailedNotFoundLock() throws TException, Inte
         .doReturn(acquiredLockResponse)
         .when(spyClient)
         .checkLock(eq(dummyLockId));
-    doNothing().when(spyOps).doUnlock(any());
     doNothing().when(spyClient).heartbeat(eq(0L), eq(dummyLockId));
 
     spyOps.doCommit(metadataV2, metadataV1);
@@ -307,7 +306,6 @@ public void testLockAcquisitionAfterFailedAndFoundLock() throws TException, Inte
 
     doReturn(showLocksResponse).when(spyClient).showLocks(any());
     doReturn(acquiredLockResponse).when(spyClient).checkLock(eq(dummyLockId));
-    doNothing().when(spyOps).doUnlock(any());
     doNothing().when(spyClient).heartbeat(eq(0L), eq(dummyLockId));
 
     spyOps.doCommit(metadataV2, metadataV1);
@@ -569,11 +567,11 @@ public void testTableLevelProcessLockBlocksConcurrentHMSRequestsForSameTable() t
   }
 
   @Test
-  public void testLockHeartbeat() throws TException {
+  public void testLockHeartbeat() throws TException, InterruptedException {
     doReturn(acquiredLockResponse).when(spyClient).lock(any());
     doAnswer(AdditionalAnswers.answersWithDelay(2000, InvocationOnMock::callRealMethod))
-        .when(spyClient)
-        .getTable(any(), any());
+        .when(spyOps)
+        .loadHmsTable();
     doNothing().when(spyClient).heartbeat(eq(0L), eq(dummyLockId));
 
     spyOps.doCommit(metadataV2, metadataV1);