@SuppressWarnings("ReverseDnsLookup")
  private LockInfo createLock() throws LockException {
    LockInfo lockInfo = new LockInfo();

    String hostName;
    try {
      hostName = InetAddress.getLocalHost().getHostName();
    } catch (UnknownHostException uhe) {
      throw new LockException(uhe, "Error generating host name");
    }

    LockComponent lockComponent =
            new LockComponent(LockType.EXCL_WRITE, LockLevel.TABLE, databaseName);
    lockComponent.setTablename(tableName);
    LockRequest lockRequest =
            new LockRequest(
                    Lists.newArrayList(lockComponent),
                    HiveHadoopUtil.currentUser(),
                    hostName);

    // Only works in Hive 2 or later.
    if (HiveVersion.min(HiveVersion.HIVE_2)) {
      lockRequest.setAgentInfo(agentInfo);
    }

    AtomicBoolean interrupted = new AtomicBoolean(false);
    Tasks.foreach(lockRequest)
            .retry(Integer.MAX_VALUE - 100)
            .exponentialBackoff(
                    lockCreationMinWaitTime, lockCreationMaxWaitTime, lockCreationTimeout, 2.0)
            .shouldRetryTest(e -> !interrupted.get() && e instanceof LockException &&
                    HiveVersion.min(HiveVersion.HIVE_2))
            .throwFailureWhenFinished()
            .run(
                request -> {
                  try {
                    LockResponse lockResponse = metaClients.run(client -> client.lock(request));
                    lockInfo.lockId = lockResponse.getLockid();
                    lockInfo.lockState = lockResponse.getState();
                  } catch (TException te) {
                    LOG.warn("Failed to create lock {}", request, te);
                    try {
                      // If we can not check for lock, or we do not find it, then rethrow the exception
                      // Otherwise we are happy as the findLock sets the lockId and the state correctly
                      if (HiveVersion.min(HiveVersion.HIVE_2)) {
                        LockInfo lockFound = findLock();
                        if (lockFound != null) {
                          lockInfo.lockId = lockFound.lockId;
                          lockInfo.lockState = lockFound.lockState;
                          LOG.info("Found lock {} by agentInfo {}", lockInfo, agentInfo);
                          return;
                        }
                      }

                      throw new LockException("Failed to find lock for table %s.%s", databaseName, tableName);
                    } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                      interrupted.set(true);
                      LOG.warn(
                              "Interrupted while trying to find lock for table {}.{}", databaseName, tableName, e);
                      throw new LockException(
                              e, "Interrupted while trying to find lock for table %s.%s", databaseName, tableName);
                    }
                  } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    interrupted.set(true);
                    LOG.warn("Interrupted while creating lock on table {}.{}", databaseName, tableName, e);
                    throw new LockException(
                            e, "Interrupted while creating lock on table %s.%s", databaseName, tableName);
                  }
                },
                LockException.class);

    // This should be initialized always, or exception should be thrown.
    LOG.debug("Lock {} created for table {}.{}", lockInfo, databaseName, tableName);
    return lockInfo;
  }
Creates a lock, retrying if possible on failure.

@return The {@link LockInfo} object for the successfully created lock
@throws LockException When we are not able to fill the hostname for lock creation, or there is
    an error during lock creation
iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/MetastoreLock.java