[HUDI-2099]hive lock which state is WATING should be released, otherwise this hive lock will be locked forever (#3186)

hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveMetastoreBasedLockProvider.java

@@ -192,6 +192,11 @@ public class HiveMetastoreBasedLockProvider implements LockProvider<LockResponse
           throw e;
         }
       }
+    } 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());
+      }
     }
   }
 

hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/functional/TestHiveMetastoreBasedLockProvider.java

@@ -127,6 +127,35 @@ public class TestHiveMetastoreBasedLockProvider extends HiveSyncFunctionalTestHa
     lockProvider.unlock();
   }
 
+  @Test
+  public void testWaitingLock() throws Exception {
+    // create different HiveMetastoreBasedLockProvider to simulate different applications
+    HiveMetastoreBasedLockProvider lockProvider1 = new HiveMetastoreBasedLockProvider(lockConfiguration, hiveConf());
+    HiveMetastoreBasedLockProvider lockProvider2 = new HiveMetastoreBasedLockProvider(lockConfiguration, hiveConf());
+    lockComponent.setOperationType(DataOperationType.NO_TXN);
+    Assertions.assertTrue(lockProvider1.acquireLock(lockConfiguration.getConfig()
+        .getLong(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP), TimeUnit.MILLISECONDS, lockComponent));
+    try {
+      boolean acquireStatus = lockProvider2.acquireLock(lockConfiguration.getConfig()
+          .getLong(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP), TimeUnit.MILLISECONDS, lockComponent);
+      Assertions.assertFalse(acquireStatus);
+    } catch (IllegalArgumentException e) {
+      // expected
+    }
+    lockProvider1.unlock();
+    // create the third HiveMetastoreBasedLockProvider to acquire lock
+    HiveMetastoreBasedLockProvider lockProvider3 = new HiveMetastoreBasedLockProvider(lockConfiguration, hiveConf());
+    boolean acquireStatus = lockProvider3.acquireLock(lockConfiguration.getConfig()
+        .getLong(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP), TimeUnit.MILLISECONDS, lockComponent);
+    // we should acquired lock, since lockProvider1 has already released lock
+    Assertions.assertTrue(acquireStatus);
+    lockProvider3.unlock();
+    // close all HiveMetastoreBasedLockProvider
+    lockProvider1.close();
+    lockProvider2.close();
+    lockProvider3.close();
+  }
+
   @Test
   public void testUnlockWithoutLock() {
     HiveMetastoreBasedLockProvider lockProvider = new HiveMetastoreBasedLockProvider(lockConfiguration, hiveConf());