@Override
  public LockHandle acquireLock(String key) throws MetaException {
    /**
     * The implementation here is a bit kludgey but done so that code exercised by unit tests
     * (which run against Derby which has no support for select for update) is as similar to
     * production code as possible.
     * In particular, with Derby we always run in a single process with a single metastore and
     * the absence of For Update is handled via a Semaphore.  The later would strictly speaking
     * make the SQL statements below unnecessary (for Derby), but then they would not be tested.
     */
    TransactionContext context = null;
    try {
      jdbcResource.bindDataSource(POOL_MUTEX);
      context = jdbcResource.getTransactionManager().getNewTransaction(PROPAGATION_REQUIRED);
      
      MapSqlParameterSource paramSource = new MapSqlParameterSource().addValue("key", key);
      String sqlStmt = sqlGenerator.addForUpdateClause("SELECT \"MT_COMMENT\", \"MT_KEY2\" FROM \"AUX_TABLE\" WHERE \"MT_KEY1\" = :key");

      LOG.debug("About to execute SQL: {}", sqlStmt);

      Long lastUpdateTime = jdbcResource.getJdbcTemplate().query(sqlStmt, paramSource, rs -> rs.next() ? rs.getLong("MT_KEY2") : null);
      if (lastUpdateTime == null) {
        try {
          jdbcResource.getJdbcTemplate().update("INSERT INTO \"AUX_TABLE\" (\"MT_KEY1\", \"MT_KEY2\") VALUES(:key, 0)", paramSource);
          context.createSavepoint();
        } catch (DataAccessException e) {
          if (!jdbcResource.getDatabaseProduct().isDuplicateKeyError(e)) {
            throw new RuntimeException("Unable to lock " + key + " due to: " + SqlRetryHandler.getMessage(e), e);
          }
          //if here, it means a concrurrent acquireLock() inserted the 'key'

          //rollback is done for the benefit of Postgres which throws (SQLState=25P02, ErrorCode=0) if
          //you attempt any stmt in a txn which had an error.
          try {
            jdbcResource.getConnection().rollback();
          } catch (SQLException ex) {
            throw new MetaException("Unable to lock " + key + " due to: " + SqlRetryHandler.getMessage(ex) + "; " + StringUtils.stringifyException(ex));
          }
        }
        lastUpdateTime = jdbcResource.getJdbcTemplate().query(sqlStmt, paramSource, rs -> rs.next() ? rs.getLong("MT_KEY2") : null);        
        if (lastUpdateTime ==null) {
          throw new IllegalStateException("Unable to lock " + key + ".  Expected row in AUX_TABLE is missing.");
        }
      }
      Semaphore derbySemaphore = null;
      if (jdbcResource.getDatabaseProduct().isDERBY()) {
        derbyKey2Lock.putIfAbsent(key, new Semaphore(1));
        derbySemaphore = derbyKey2Lock.get(key);
        derbySemaphore.acquire();
      }
      if (LOG.isDebugEnabled()) {
        LOG.debug("{} locked by {}", key, JavaUtils.hostname());
      }
      //OK, so now we have a lock
      return new LockHandleImpl(jdbcResource, context, key, lastUpdateTime, derbySemaphore);
    } catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
      if (context != null) {
        jdbcResource.getTransactionManager().rollback(context);
      }
      throw new MetaException("Unable to lock " + key + " due to: " + ex.getMessage() + StringUtils.stringifyException(ex));
    } catch (Throwable e) {
      if (context != null) {
        jdbcResource.getTransactionManager().rollback(context);
      }
      throw e;
    } finally {
      jdbcResource.unbindDataSource();
    }
  }
must be static since even in UT there may be > 1 instance of TxnHandler
(e.g. via Compactor services)
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnStoreMutex.java