@Test
  public void testTableLevelProcessLockBlocksConcurrentHMSRequestsForSameTable() throws Exception {
    int numConcurrentCommits = 10;
    // resetting the spy client to forget about prior call history
    reset(spyClient);

    // simulate several concurrent commit operations on the same table
    ExecutorService executor = Executors.newFixedThreadPool(numConcurrentCommits);
    IntStream.range(0, numConcurrentCommits).forEach(i ->
        executor.submit(() -> {
          try {
            spyOps.doCommit(metadataV2, metadataV1);
          } catch (CommitFailedException e) {
            // failures are expected here when checking the base version
            // it's no problem, we're not testing the actual commit success here, only the HMS lock acquisition attempts
          }
        }));
    executor.shutdown();
    executor.awaitTermination(30, TimeUnit.SECONDS);

    // intra-process commits to the same table should be serialized now
    // i.e. no thread should receive WAITING state from HMS and have to call checkLock periodically
    verify(spyClient, never()).checkLock(any(Long.class));
    // all threads eventually got their turn
    verify(spyClient, times(numConcurrentCommits)).lock(any(LockRequest.class));
  }
Wraps an ArgumentCaptor to provide data based on the request
iceberg/iceberg-catalog/src/test/java/org/apache/iceberg/hive/TestHiveCommitLocks.java