code: public long lockBlock(long sessionId, long blockId, BlockLockType blockLockType) {
      ClientRWLock blockLock = getBlockLock(blockId);
      Lock lock;
      if (blockLockType == BlockLockType.READ) {
        lock = blockLock.readLock();
      } else {
        // Make sure the session isn't already holding the block lock.
        if (sessionHoldsLock(sessionId, blockId)) {
          throw new IllegalStateException(String
              .format("Session %s attempted to take a write lock on block %s, but the session already"
                  + " holds a lock on the block", sessionId, blockId));
        }
        lock = blockLock.writeLock();
      }
      lock.lock();
      try {
        long lockId = LOCK_ID_GEN.getAndIncrement();
        synchronized (mSharedMapsLock) {
          mLockIdToRecordMap.put(lockId, new LockRecord(sessionId, blockId, lock));
          Set<Long> sessionLockIds = mSessionIdToLockIdsMap.get(sessionId);
          if (sessionLockIds == null) {
            mSessionIdToLockIdsMap.put(sessionId, Sets.newHashSet(lockId));
          } else {
            sessionLockIds.add(lockId);
          }
        }
        return lockId;
      } catch (RuntimeException e) {
        // If an unexpected exception occurs, we should release the lock to be conservative.
        unlock(lock, blockId);
        throw e;
      }
    }
docs: Locks a block. Note that even if this block does not exist, a lock id is still returned.
  
  If all {@link PropertyKey#WORKER_TIERED_STORE_BLOCK_LOCKS} are already in use and no lock has
  been allocated for the specified block, this method will need to wait until a lock can be
  acquired from the lock pool.
  
  @param sessionId the session id
  @param blockId the block id
  @param blockLockType {@link BlockLockType#READ} or {@link BlockLockType#WRITE}
  @return lock id