private MoveBlockResult moveBlockInternal(long sessionId, long blockId,
      BlockStoreLocation oldLocation, BlockStoreLocation newLocation)
          throws BlockDoesNotExistException, BlockAlreadyExistsException,
          InvalidWorkerStateException, IOException {
    long lockId = mLockManager.lockBlock(sessionId, blockId, BlockLockType.WRITE);
    try {
      long blockSize;
      String srcFilePath;
      String dstFilePath;
      BlockMeta srcBlockMeta;
      BlockStoreLocation srcLocation;
      BlockStoreLocation dstLocation;

      try (LockResource r = new LockResource(mMetadataReadLock)) {
        if (mMetaManager.hasTempBlockMeta(blockId)) {
          throw new InvalidWorkerStateException(ExceptionMessage.MOVE_UNCOMMITTED_BLOCK, blockId);
        }
        srcBlockMeta = mMetaManager.getBlockMeta(blockId);
        srcLocation = srcBlockMeta.getBlockLocation();
        srcFilePath = srcBlockMeta.getPath();
        blockSize = srcBlockMeta.getBlockSize();
      }

      if (!srcLocation.belongsTo(oldLocation)) {
        throw new BlockDoesNotExistException(ExceptionMessage.BLOCK_NOT_FOUND_AT_LOCATION, blockId,
            oldLocation);
      }
      TempBlockMeta dstTempBlock =
          createBlockMetaInternal(sessionId, blockId, newLocation, blockSize, false);
      if (dstTempBlock == null) {
        return new MoveBlockResult(false, blockSize, null, null);
      }

      // When `newLocation` is some specific location, the `newLocation` and the `dstLocation` are
      // just the same; while for `newLocation` with a wildcard significance, the `dstLocation`
      // is a specific one with specific tier and dir which belongs to newLocation.
      dstLocation = dstTempBlock.getBlockLocation();

      // When the dstLocation belongs to srcLocation, simply abort the tempBlockMeta just created
      // internally from the newLocation and return success with specific block location.
      if (dstLocation.belongsTo(srcLocation)) {
        mMetaManager.abortTempBlockMeta(dstTempBlock);
        return new MoveBlockResult(true, blockSize, srcLocation, dstLocation);
      }
      dstFilePath = dstTempBlock.getCommitPath();

      // Heavy IO is guarded by block lock but not metadata lock. This may throw IOException.
      FileUtils.move(srcFilePath, dstFilePath);

      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        // If this metadata update fails, we panic for now.
        // TODO(bin): Implement rollback scheme to recover from IO failures.
        mMetaManager.moveBlockMeta(srcBlockMeta, dstTempBlock);
      } catch (BlockAlreadyExistsException | BlockDoesNotExistException
          | WorkerOutOfSpaceException e) {
        // WorkerOutOfSpaceException is only possible if session id gets cleaned between
        // createBlockMetaInternal and moveBlockMeta.
        throw Throwables.propagate(e); // we shall never reach here
      }

      return new MoveBlockResult(true, blockSize, srcLocation, dstLocation);
    } finally {
      mLockManager.unlockBlock(lockId);
    }
  }
private MoveBlockResult moveBlockInternal ( long sessionId , long blockId , BlockStoreLocation oldLocation , BlockStoreLocation newLocation ) throws BlockDoesNotExistException , BlockAlreadyExistsException , InvalidWorkerStateException , IOException { long lockId = mLockManager . lockBlock ( sessionId , blockId , BlockLockType . WRITE ) ; try { long blockSize ; String srcFilePath ; String dstFilePath ; BlockMeta srcBlockMeta ; BlockStoreLocation srcLocation ; BlockStoreLocation dstLocation ; try ( LockResource r = new LockResource ( mMetadataReadLock ) ) { if ( mMetaManager . hasTempBlockMeta ( blockId ) ) { throw new InvalidWorkerStateException ( ExceptionMessage . MOVE_UNCOMMITTED_BLOCK , blockId ) ; } srcBlockMeta = mMetaManager . getBlockMeta ( blockId ) ; srcLocation = srcBlockMeta . getBlockLocation ( ) ; srcFilePath = srcBlockMeta . getPath ( ) ; blockSize = srcBlockMeta . getBlockSize ( ) ; } if ( ! srcLocation . belongsTo ( oldLocation ) ) { throw new BlockDoesNotExistException ( ExceptionMessage . BLOCK_NOT_FOUND_AT_LOCATION , blockId , oldLocation ) ; } TempBlockMeta dstTempBlock = createBlockMetaInternal ( sessionId , blockId , newLocation , blockSize , false ) ; if ( dstTempBlock == null ) { return new MoveBlockResult ( false , blockSize , null , null ) ; } dstLocation = dstTempBlock . getBlockLocation ( ) ; if ( dstLocation . belongsTo ( srcLocation ) ) { mMetaManager . abortTempBlockMeta ( dstTempBlock ) ; return new MoveBlockResult ( true , blockSize , srcLocation , dstLocation ) ; } dstFilePath = dstTempBlock . getCommitPath ( ) ; FileUtils . move ( srcFilePath , dstFilePath ) ; try ( LockResource r = new LockResource ( mMetadataWriteLock ) ) { mMetaManager . moveBlockMeta ( srcBlockMeta , dstTempBlock ) ; } catch ( BlockAlreadyExistsException | BlockDoesNotExistException | WorkerOutOfSpaceException e ) { throw Throwables . propagate ( e ) ; } return new MoveBlockResult ( true , blockSize , srcLocation , dstLocation ) ; } finally { mLockManager . unlockBlock ( lockId ) ; } }
Moves a block to new location only if allocator finds available space in newLocation. This
method will not trigger any eviction. Returns {@link MoveBlockResult}.

@param sessionId session id
@param blockId block id
@param oldLocation the source location of the block
@param newLocation new location to move this block
@return the resulting information about the move operation
@throws BlockDoesNotExistException if block is not found
@throws BlockAlreadyExistsException if a block with same id already exists in new location
@throws InvalidWorkerStateException if the block to move is a temp block