void initializeHeldLocks(InternalDistributedMember owner, Set tokens)
      throws InterruptedException {
    synchronized (this) {
      if (isDestroyed()) {
        return;
      }
      if (!acquireDestroyReadLock(0)) {
        return;
      }
    }

    try {
      synchronized (grantTokens) {
        Set members = dlock.getDistributionManager().getDistributionManagerIds();

        final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE);
        for (final Object o : tokens) {
          DLockRemoteToken token = (DLockRemoteToken) o;
          DLockGrantToken grantToken = getOrCreateGrant(token.getName());
          try {

            // make sure the token's owner is still in the system
            if (!members.contains(owner)) {
              // skipping because member is no longer in view
              if (isDebugEnabled_DLS) {
                logger.trace(LogMarker.DLS_VERBOSE,
                    "Initialization of held locks is skipping {} because owner {} is not in view: {}",
                    token, owner, members);
              }
              continue;
            }

            RemoteThread rThread = null;
            boolean isSuspendLock = false;
            int lockId = -1;

            synchronized (grantToken) {
              if (grantToken.isLeaseHeld()) {
                logger.error(LogMarker.DLS_MARKER,
                    "Initialization of held locks is skipping {} because lock is already held: {}",
                    token, grantToken);
                continue;
              }

              grantToken.grantLock(owner, token.getLeaseExpireTime(), token.getLeaseId(),
                  token.getLesseeThread());

              // grantToken may have already expired or is about to expire
              // complete initialization but make sure grantor thread will wake
              // up and expire it as soon as it's running
              if (grantToken.getLeaseExpireTime() > -1
                  && grantToken.getLeaseExpireTime() < Long.MAX_VALUE) {
                long now = DLockService.getLockTimeStamp(dm);
                thread.checkTimeToWait(grantToken.getLeaseExpireTime() - now, true);
              }

              rThread = grantToken.getRemoteThread();
              isSuspendLock = grantToken.isSuspendLockingToken();
              lockId = grantToken.getLockId();
            }

            // update the readLock and suspendLocking states...
            synchronized (suspendLock) {
              if (isSuspendLock) {
                suspendLocking(rThread, lockId);
              } else {
                Assert.assertTrue(!isLockingSuspended() || isLockingSuspendedBy(rThread),
                    "Locking is suspended by a different thread: " + token);
                Integer integer = (Integer) readLockCountMap.get(rThread);
                int readLockCount = integer == null ? 0 : integer;
                readLockCount++;
                readLockCountMap.put(rThread, readLockCount);
                totalReadLockCount++;
                checkTotalReadLockCount();
              }
            } // suspendLock sync

          } finally {
            grantToken.decAccess();
          }

        } // tokens iter
      } // grantTokens sync
      return;
    } finally {
      releaseDestroyReadLock();
    }
  }
Initializes this new grantor with previously held locks as provided during grantor recovery.
<p>
Acquires destroyReadLock. Synchronizes on this grantor, grantTokens, suspendLock, the grant
token.

@param owner the member that owns the tokens to be scheduled
@param tokens set of DLockRemoteTokens to be scheduled for owner
geode-core/src/main/java/org/apache/geode/distributed/internal/locks/DLockGrantor.java