public boolean lockInterruptibly(final Object name, final long waitTimeMillis,
      final long leaseTimeMillis, final boolean tryLock, final boolean interruptible,
      final boolean disallowReentrant, final boolean disableAlerts) throws InterruptedException {
    checkDestroyed();

    boolean interrupted = Thread.interrupted();
    if (interrupted && interruptible) {
      throw new InterruptedException();
    }

    try {
      long statStart = getStats().startLockWait();
      long startTime = getLockTimeStamp(dm);

      long requestWaitTime = waitTimeMillis;
      long requestLeaseTime = leaseTimeMillis;

      // -1 means "lease forever". Long.MAX_VALUE is pretty close.
      if (requestLeaseTime == -1) {
        requestLeaseTime = Long.MAX_VALUE;
      }
      // -1 means "wait forever". Long.MAX_VALUE is pretty close.
      if (requestWaitTime == -1) {
        requestWaitTime = Long.MAX_VALUE;
      }

      long waitLimit = startTime + requestWaitTime;
      if (waitLimit < 0) {
        waitLimit = Long.MAX_VALUE;
      }

      if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
        logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - entering lock()", this, name);
      }

      DLockToken token = getOrCreateToken(name);
      boolean gotLock = false;
      blockedOn.set(name);
      try { // try-block for end stats, token cleanup, and interrupt check

        ThreadRequestState requestState = threadRequestState.get();
        if (requestState == null) {
          requestState = new ThreadRequestState(incThreadSequence(), interruptible);
          threadRequestState.set(requestState);
        } else {
          requestState.interruptible = interruptible;
        }
        final int threadId = requestState.threadId;

        // if reentry and no change to expiration then grantor is not bothered

        boolean keepTrying = true;
        int lockId = -1;
        incActiveLocks();

        while (keepTrying) {
          checkDestroyed();
          interrupted = Thread.interrupted() || interrupted; // clear
          if (interrupted && interruptible) {
            throw new InterruptedException();
          }

          // Check for recursive lock
          boolean reentrant = false;
          int recursionBefore = -1;

          synchronized (token) {
            token.checkForExpiration();
            if (token.isLeaseHeldByCurrentThread()) {
              if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
                logger.trace(LogMarker.DLS_VERBOSE, "{} , name: {} - lock() is reentrant: {}", this,
                    name, token);
              }
              reentrant = true;
              if (disallowReentrant) {
                throw new IllegalStateException(
                    String.format("%s attempted to reenter non-reentrant lock %s",
                        Thread.currentThread(), token));
              }
              recursionBefore = token.getRecursion();
              lockId = token.getLeaseId(); // keep lockId
              if (lockId < 0) {
                // loop back around due to expiration
                continue;
              }
            } // isLeaseHeldByCurrentThread
          } // token sync

          LockGrantorId theLockGrantorId = getLockGrantorId();

          if (reentrant) {
            Assert.assertTrue(lockId > -1, "Reentrant lock must have lockId > -1");
            // lockId = token.getLockId(); // keep lockId
          } else {
            // this thread is not current owner...
            lockId = -1; // reset lockId back to -1
          }

          DLockRequestProcessor processor = createRequestProcessor(theLockGrantorId, name, threadId,
              startTime, requestLeaseTime, requestWaitTime, reentrant, tryLock, disableAlerts);
          if (reentrant) {
            // check for race condition... reentrant expired already...
            // related to bug 32765, but client-side... see bug 33402
            synchronized (token) {
              if (!token.isLeaseHeldByCurrentThread()) {
                reentrant = false;
                recursionBefore = -1;
                token.checkForExpiration();
              }
            }
          } else {
            // set lockId since this is the first granting (non-reentrant)
            lockId = processor.getProcessorId();
          }

          gotLock = processor.requestLock(interruptible, lockId); // can throw
                                                                  // InterruptedException

          if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
            logger.trace(LogMarker.DLS_VERBOSE, "Grantor {} replied {}", theLockGrantorId,
                processor.getResponseCodeString());
          }

          if (gotLock) {
            final long leaseExpireTime = processor.getLeaseExpireTime();
            int recursion = recursionBefore + 1;

            if (!grantLocalDLockAfterObtainingRemoteLock(name, token, threadId, leaseExpireTime,
                lockId, theLockGrantorId, processor, recursion)) {
              continue;
            }

            if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
              logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - granted lock: {}", this, name,
                  token);
            }
            keepTrying = false;
          } else if (processor.repliedDestroyed()) {
            checkDestroyed(); // throws LockServiceDestroyedException
            Assert.assertTrue(isDestroyed(),
                "Grantor reports service " + this + " is destroyed: " + name);
          } else if (processor.repliedNotGrantor() || processor.hadNoResponse()) {
            long waitForGrantorTime = waitLimit - token.getCurrentTime();
            if (waitForGrantorTime <= 0) {
              waitForGrantorTime = 100;
            }
            notLockGrantorId(theLockGrantorId, waitForGrantorTime, TimeUnit.MILLISECONDS);
            // keepTrying is still true... loop back around
          } else if (processor.repliedNotHolder()) {
            // fix part of bug 32765 - reentrant/expiration problem
            // probably expired... try to get non-reentrant lock
            reentrant = false;
            recursionBefore = -1;
            synchronized (token) {
              token.checkForExpiration();
              if (token.isLeaseHeldByCurrentThread()) {
                // THIS SHOULDN'T HAPPEN -- some sort of weird consistency
                // problem. Do what the grantor says and release the lock...
                logger.warn(LogMarker.DLS_MARKER, "Grantor reports reentrant lock not held: {}",
                    token);

                // Attempt at fault tolerance: We thought we owned it, but we
                // don't; let's release it. Removes hot loop in bug 37276,
                // but does not address underlying consistency failure.
                RemoteThread rThread = new RemoteThread(getDistributionManager().getId(), threadId);
                token.releaseLock(lockId, rThread, false);
              }
            } // token sync
          } // grantor replied NOT_HOLDER for reentrant lock

          else {
            // either dlock service is suspended or tryLock failed
            // fixed the math here... bug 32765
            if (waitLimit > token.getCurrentTime() + 20) {
              sleep(20, interruptible);
            }
            keepTrying = waitLimit > token.getCurrentTime();
          }

        } // while (keepTrying)
          // try-block for end stats, token cleanup, and interrupt check
      } finally {
        getStats().endLockWait(statStart, gotLock);

        // cleanup token if failed to get lock
        if (!gotLock) {
          synchronized (token) {
            token.decUsage();
          }
          freeResources(token.getName());
        }

        // reset the interrupt state
        if (interrupted) {
          Thread.currentThread().interrupt();
        }

        // throw InterruptedException only if failed to get lock and interrupted
        if (!gotLock && interruptible && Thread.interrupted()) {
          throw new InterruptedException();
        }
        blockedOn.set(null);
      }

      if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
        logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - exiting lock() returning {}", this,
            name, gotLock);
      }
      return gotLock;
    } finally {
      if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) {
        logger.trace(LogMarker.DLS_VERBOSE, "{}, name: {} - exiting lock() without returning value",
            this, name);
      }
      if (interrupted) {
        Thread.currentThread().interrupt();
      }
    }
  }
@param name the name of the lock to acquire in this service. This object must conform to the
       general contract of <code>equals(Object)</code> and <code>hashCode()</code> as described
       in {@link java.lang.Object#hashCode()}.

@param waitTimeMillis the number of milliseconds to try to acquire the lock before giving up
       and returning false. A value of -1 causes this method to block until the lock is
       acquired.

@param leaseTimeMillis the number of milliseconds to hold the lock after granting it, before
       automatically releasing it if it hasn't already been released by invoking
       {@link #unlock(Object)}. If <code>leaseTimeMillis</code> is -1, hold the lock until
       explicitly unlocked.

@param tryLock true if the lock should be acquired or fail if currently held. waitTimeMillis
       will be ignored if the lock is currently held by another client.

@param interruptible true if this lock request is interruptible

@param disableAlerts true to disable logging alerts if the dlock is taking a long time to
       acquired.

@return true if the lock was acquired, false if the timeout <code>waitTimeMillis</code> passed
        without acquiring the lock.

@throws InterruptedException if the thread is interrupted before or during this method.

@throws UnsupportedOperationException if attempt to lock batch involves non-tryLocks
geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockService.java