alluxio-master-2.x/core/server/worker/src/test/java/alluxio/worker/block/TieredBlockStoreTest.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.worker.block;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

import alluxio.Constants;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.runtime.ResourceExhaustedRuntimeException;
import alluxio.retry.CountingRetry;
import alluxio.retry.RetryPolicy;
import alluxio.test.util.ConcurrencyUtils;
import alluxio.util.CommonUtils;
import alluxio.util.io.FileUtils;
import alluxio.worker.block.annotator.BlockIterator;
import alluxio.worker.block.evictor.EvictionPlan;
import alluxio.worker.block.evictor.Evictor;
import alluxio.worker.block.evictor.Evictor.Mode;
import alluxio.worker.block.meta.DefaultBlockMeta;
import alluxio.worker.block.meta.DefaultTempBlockMeta;
import alluxio.worker.block.meta.StorageDir;
import alluxio.worker.block.meta.TempBlockMeta;

import com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.mockito.invocation.InvocationOnMock;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * Unit tests for {@link TieredBlockStore}.
 */
public final class TieredBlockStoreTest {
  private static final long SESSION_ID1 = 2;
  private static final long SESSION_ID2 = 3;
  private static final long BLOCK_ID1 = 1000;
  private static final long BLOCK_ID2 = 1001;
  private static final long TEMP_BLOCK_ID = 1003;
  private static final long TEMP_BLOCK_ID2 = 1004;
  private static final long BLOCK_SIZE = 512;
  private static final String FIRST_TIER_ALIAS = TieredBlockStoreTestUtils.TIER_ALIAS[0];
  private static final String SECOND_TIER_ALIAS = TieredBlockStoreTestUtils.TIER_ALIAS[1];
  private TieredBlockStore mBlockStore;
  private BlockMetadataManager mMetaManager;
  private BlockLockManager mLockManager;
  private StorageDir mTestDir1;
  private StorageDir mTestDir2;
  private StorageDir mTestDir3;
  private StorageDir mTestDir4;
  private BlockIterator mBlockIterator;

  /** Rule to create a new temporary folder during each test. */
  @Rule
  public TemporaryFolder mTestFolder = new TemporaryFolder();

  /** The exception expected to be thrown. */
  @Rule
  public ExpectedException mThrown = ExpectedException.none();

  @Before
  public void before() throws Exception {
    Configuration.reloadProperties();
    init(0);
  }

  private void init(long reservedBytes) throws Exception {
    Configuration.set(PropertyKey.WORKER_REVIEWER_CLASS,
            "alluxio.worker.block.reviewer.AcceptingReviewer");
    // No reserved space for tests that are not swap related.
    Configuration.set(PropertyKey.WORKER_MANAGEMENT_TIER_ALIGN_RESERVED_BYTES, reservedBytes);

    File tempFolder = mTestFolder.newFolder();
    TieredBlockStoreTestUtils.setupDefaultConf(tempFolder.getAbsolutePath());
    mMetaManager = BlockMetadataManager.createBlockMetadataManager();
    mLockManager = new BlockLockManager();
    mBlockStore = new TieredBlockStore(mMetaManager, mLockManager);
    mBlockIterator = mMetaManager.getBlockIterator();

    mTestDir1 = mMetaManager.getTier(FIRST_TIER_ALIAS).getDir(0);
    mTestDir2 = mMetaManager.getTier(FIRST_TIER_ALIAS).getDir(1);
    mTestDir3 = mMetaManager.getTier(SECOND_TIER_ALIAS).getDir(1);
    mTestDir4 = mMetaManager.getTier(SECOND_TIER_ALIAS).getDir(2);
  }

  @Test
  public void differentSessionLockDifferentBlocks() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, BLOCK_SIZE, mTestDir2, mMetaManager,
        mBlockIterator);

    Optional<BlockLock> lock1 = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID1);
    assertTrue(
        Sets.difference(mLockManager.getLockedBlocks(), Sets.newHashSet(BLOCK_ID1)).isEmpty());

    Optional<BlockLock> lock2 = mBlockStore.pinBlock(SESSION_ID2, BLOCK_ID2);
    assertNotEquals(lock1, lock2);
    assertTrue(
        Sets.difference(mLockManager.getLockedBlocks(), Sets.newHashSet(BLOCK_ID1, BLOCK_ID2))
            .isEmpty());

    lock2.get().close();
    assertTrue(
        Sets.difference(mLockManager.getLockedBlocks(), Sets.newHashSet(BLOCK_ID1)).isEmpty());

    lock1.get().close();
    assertTrue(mLockManager.getLockedBlocks().isEmpty());
  }

  @Test
  public void sameSessionLockDifferentBlocks() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID2, BLOCK_SIZE, mTestDir2, mMetaManager,
        mBlockIterator);

    BlockLock lockId1 = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID1).get();
    assertTrue(
        Sets.difference(mLockManager.getLockedBlocks(), Sets.newHashSet(BLOCK_ID1)).isEmpty());

    BlockLock lockId2 = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID2).get();
    assertNotEquals(lockId1.get(), lockId2.get());
  }

  @Test
  public void pinNonExistingBlock() {
    assertFalse(mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID1).isPresent());
  }

  @Test
  public void commitBlock() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    assertFalse(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    mBlockStore.commitBlock(SESSION_ID1, TEMP_BLOCK_ID, false);
    assertTrue(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    assertFalse(
        FileUtils.exists(DefaultTempBlockMeta.tempPath(mTestDir1, SESSION_ID1, TEMP_BLOCK_ID)));
    assertTrue(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, TEMP_BLOCK_ID)));
  }

  @Test
  public void abortBlock() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.abortBlock(SESSION_ID1, TEMP_BLOCK_ID);
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    assertFalse(
        FileUtils.exists(DefaultTempBlockMeta.tempPath(mTestDir1, SESSION_ID1, TEMP_BLOCK_ID)));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, TEMP_BLOCK_ID)));
  }

  @Test
  public void moveBlock() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
        AllocateOptions.forMove(mTestDir2.toBlockStoreLocation()));
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertTrue(mTestDir2.hasBlockMeta(BLOCK_ID1));
    assertTrue(mBlockStore.hasBlockMeta(BLOCK_ID1));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));
    assertTrue(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir2, BLOCK_ID1)));

    // Move block from the specific Dir
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    // Move block from right Dir
    mBlockStore.moveBlock(SESSION_ID2, BLOCK_ID2,
        AllocateOptions.forMove(mTestDir3.toBlockStoreLocation()));
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID2));
    assertTrue(mTestDir3.hasBlockMeta(BLOCK_ID2));
    assertTrue(mBlockStore.hasBlockMeta(BLOCK_ID2));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID2)));
    assertTrue(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir3, BLOCK_ID2)));
  }

  @Test
  public void tierMoveTargetIsFull() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
            mBlockIterator);
    // Fill up the target dir
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, mTestDir2.getCapacityBytes(),
            mTestDir2, mMetaManager, mBlockIterator);
    mThrown.expect(ResourceExhaustedRuntimeException.class);
    mThrown.expectMessage(ExceptionMessage.NO_SPACE_FOR_BLOCK_MOVE.getMessage(
            mTestDir2.toBlockStoreLocation(), BLOCK_ID1));
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
            AllocateOptions.forTierMove(mTestDir2.toBlockStoreLocation()));
    // Consistency check: the block is still in its original location
    assertTrue(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(mTestDir2.hasBlockMeta(BLOCK_ID1));
    assertTrue(mBlockStore.hasBlockMeta(BLOCK_ID1));
    assertTrue(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir2, BLOCK_ID1)));
  }

  @Test
  public void moveBlockToSameLocationDoesNothing() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    // Move block to same location will simply do nothing, so the src block keeps where it was,
    // and the available space should also remain unchanged.
    long availableBytesBefore = mMetaManager.getAvailableBytes(mTestDir1.toBlockStoreLocation());
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
        AllocateOptions.forMove(mTestDir1.toBlockStoreLocation()));
    long availableBytesAfter = mMetaManager.getAvailableBytes(mTestDir1.toBlockStoreLocation());

    assertEquals(availableBytesBefore, availableBytesAfter);
    assertTrue(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(mMetaManager.hasTempBlockMeta(BLOCK_ID1));
    assertTrue(mBlockStore.hasBlockMeta(BLOCK_ID1));
    assertTrue(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));
  }

  /**
   * Tests the {@link TieredBlockStore#removeBlock(long, long)} method.
   */
  @Test
  public void removeBlock() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mBlockStore.removeBlock(SESSION_ID1, BLOCK_ID1);
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(mBlockStore.hasBlockMeta(BLOCK_ID1));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));

    // Remove block from specific Dir
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    // Remove block from right Dir
    mBlockStore.removeBlock(SESSION_ID2, BLOCK_ID2);
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID2));
    assertFalse(mBlockStore.hasBlockMeta(BLOCK_ID2));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID2)));

    // Remove block from the specific tier
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mBlockStore.removeBlock(SESSION_ID2, BLOCK_ID2);
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID2));
    assertFalse(mBlockStore.hasBlockMeta(BLOCK_ID2));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID2)));
  }

  @Test
  public void freeSpace() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mBlockStore.freeSpace(SESSION_ID1, mTestDir1.getCapacityBytes(), mTestDir1.getCapacityBytes(),
        mTestDir1.toBlockStoreLocation());
    // Expect BLOCK_ID1 to be moved out of mTestDir1
    assertEquals(mTestDir1.getCapacityBytes(), mTestDir1.getAvailableBytes());
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));
  }

  @Test
  public void freeSpaceAhead() throws Exception {
    long halfCapacityBytes = mTestDir1.getCapacityBytes() / 2;
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, halfCapacityBytes, mTestDir1,
        mMetaManager, mBlockIterator);
    TieredBlockStoreTestUtils.cache2(SESSION_ID2, BLOCK_ID2, halfCapacityBytes, mTestDir1,
        mMetaManager, mBlockIterator);
    mBlockStore.freeSpace(SESSION_ID1, halfCapacityBytes, 2 * halfCapacityBytes,
        mTestDir1.toBlockStoreLocation());
    // Expect BLOCK_ID1 and BLOCK_ID2 to be moved out of mTestDir1
    assertEquals(mTestDir1.getCapacityBytes(), mTestDir1.getAvailableBytes());
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID1));
    assertFalse(mTestDir1.hasBlockMeta(BLOCK_ID2));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID1)));
    assertFalse(FileUtils.exists(DefaultBlockMeta.commitPath(mTestDir1, BLOCK_ID2)));
  }

  @Test
  public void freeSpaceThreadSafe() throws Exception {
    int threadAmount = 10;
    List<Runnable> runnables = new ArrayList<>();
    Evictor evictor = mock(Evictor.class);
    when(
        evictor.freeSpaceWithView(any(Long.class), any(BlockStoreLocation.class),
            any(BlockMetadataEvictorView.class), any(Mode.class)))
        .thenAnswer((InvocationOnMock invocation) -> {
          CommonUtils.sleepMs(20);
          return new EvictionPlan(new ArrayList<>(), new ArrayList<>());
        }
        );
    for (int i = 0; i < threadAmount; i++) {
      runnables.add(() -> {
        try {
          mBlockStore.freeSpace(SESSION_ID1, 0, 0,
              new BlockStoreLocation(Constants.MEDIUM_MEM, 0));
        } catch (Exception e) {
          fail();
        }
      });
    }
    RetryPolicy retry = new CountingRetry(threadAmount);
    while (retry.attempt()) {
      ConcurrencyUtils.assertConcurrent(runnables, threadAmount);
    }
  }

  @Test
  public void freeSpaceWithPinnedBlocks() throws Exception {
    // create a pinned block
    TieredBlockStoreTestUtils.cache(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mBlockStore,
        mTestDir1.toBlockStoreLocation(), true);

    // Expect an empty eviction plan is feasible
    mThrown.expect(ResourceExhaustedRuntimeException.class);
    mThrown.expectMessage(ExceptionMessage.NO_EVICTION_PLAN_TO_FREE_SPACE.getMessage(
        mTestDir1.getCapacityBytes(), mTestDir1.toBlockStoreLocation().tierAlias()));
    mBlockStore.freeSpace(SESSION_ID1, mTestDir1.getCapacityBytes(), mTestDir1.getCapacityBytes(),
        mTestDir1.toBlockStoreLocation());
  }

  @Test
  public void requestSpace() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, 1, mTestDir1);
    mBlockStore.requestSpace(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE - 1);
    assertTrue(mTestDir1.hasTempBlockMeta(TEMP_BLOCK_ID));
    assertEquals(BLOCK_SIZE, mTestDir1.getTempBlockMeta(TEMP_BLOCK_ID).get().getBlockSize());
    assertEquals(mTestDir1.getCapacityBytes() - BLOCK_SIZE, mTestDir1.getAvailableBytes());
  }

  @Test
  public void requestSpaceFailedWhenStorageFull() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, 1, mTestDir1);
    // Fill up the target dir
    TieredBlockStoreTestUtils.cache(SESSION_ID1, BLOCK_ID1, mTestDir1.getAvailableBytes(),
        mBlockStore, mTestDir1.toBlockStoreLocation(), true);
    TieredBlockStoreTestUtils.cache(SESSION_ID2, BLOCK_ID2, mTestDir2.getAvailableBytes(),
        mBlockStore, mTestDir2.toBlockStoreLocation(), true);
    Assert.assertThrows(ResourceExhaustedRuntimeException.class,
        () -> mBlockStore.requestSpace(SESSION_ID1, TEMP_BLOCK_ID, 1));
  }

  @Test
  public void requestSpaceFailed() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID,
        mTestDir1.getAvailableBytes(), mTestDir1);
    Assert.assertThrows(ResourceExhaustedRuntimeException.class,
        () -> mBlockStore.requestSpace(SESSION_ID1, TEMP_BLOCK_ID, 1));
  }

  @Test
  public void createBlockMetaWithoutEviction() {
    TempBlockMeta tempBlockMeta = mBlockStore.createBlock(SESSION_ID1, TEMP_BLOCK_ID,
        AllocateOptions.forCreate(1, mTestDir1.toBlockStoreLocation()));
    assertEquals(1, tempBlockMeta.getBlockSize());
    assertEquals(mTestDir1, tempBlockMeta.getParentDir());
  }

  @Test
  public void createBlockMetaWithMediumType() {
    BlockStoreLocation loc = BlockStoreLocation.anyDirInAnyTierWithMedium(Constants.MEDIUM_MEM);
    TempBlockMeta tempBlockMeta = mBlockStore.createBlock(SESSION_ID1, TEMP_BLOCK_ID,
        AllocateOptions.forCreate(1, loc));
    assertEquals(1, tempBlockMeta.getBlockSize());
    assertEquals(mTestDir2, tempBlockMeta.getParentDir());

    BlockStoreLocation loc2 = BlockStoreLocation.anyDirInAnyTierWithMedium(Constants.MEDIUM_SSD);
    TempBlockMeta tempBlockMeta2 = mBlockStore.createBlock(SESSION_ID1, TEMP_BLOCK_ID2,
        AllocateOptions.forCreate(1, loc2));
    assertEquals(1, tempBlockMeta2.getBlockSize());
    assertEquals(mTestDir4, tempBlockMeta2.getParentDir());
  }

  /**
   * Tests that when creating a block, if the space of the target location is currently taken by
   * another block being locked, this creation will happen on the next available dir.
   */
  @Test
  public void createBlockMetaWithBlockLocked() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);

    // session1 locks a block first
    BlockLock lock = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID1).get();

    // Create file that in dir1 that won't fit.
    TieredBlockStoreTestUtils.cache(SESSION_ID2, TEMP_BLOCK_ID, mTestDir1.getCapacityBytes(),
        mBlockStore, mTestDir1.toBlockStoreLocation(), false);

    // unlock the original block.
    lock.close();

    assertEquals(mTestDir1.getCapacityBytes(), mTestDir2.getCommittedBytes());
  }

  @Test
  public void relaxLocationWrites() throws Exception {
    long totalCapacityBytes = 0;
    for (StorageDir dir : new StorageDir[] {mTestDir1, mTestDir2, mTestDir3, mTestDir4}) {
      totalCapacityBytes += dir.getCapacityBytes();
    }

    long fileSizeBytes = 1000;
    int maxFileCount = (int) (totalCapacityBytes / fileSizeBytes);

    // Write to first dir, 2 times of the whole store's peak capacity.
    for (int i = 0; i <= maxFileCount * 2; i++) {
      TieredBlockStoreTestUtils.cache(i, i, fileSizeBytes, mBlockStore,
          mTestDir1.toBlockStoreLocation(), false);
    }
  }

  @Test
  public void stickyLocationWrites() throws Exception {
    long fileSizeBytes = 100;
    int maxFileCount = (int) (mTestDir1.getCapacityBytes() / fileSizeBytes);
    AllocateOptions remainOnLocOptions = AllocateOptions
        .forCreate(fileSizeBytes, mTestDir1.toBlockStoreLocation()).setForceLocation(true);

    // Write to first dir, 2 times of its peak capacity.
    for (int i = 0; i <= maxFileCount * 2; i++) {
      TieredBlockStoreTestUtils.cache(i, i, remainOnLocOptions, mBlockStore, false);
    }

    assertEquals(0, mTestDir1.getAvailableBytes());
    assertEquals(0, mTestDir2.getCommittedBytes());
    assertEquals(0, mTestDir3.getCommittedBytes());
    assertEquals(0, mTestDir4.getCommittedBytes());
  }

  /**
   * Tests that when moving a block from src location to dst, if the space of the dst location is
   * currently taken by another block being locked, this move operation will fail until the lock
   * released.
   */
  @Test
  public void moveBlockMetaWithBlockLocked() throws Exception {
    // Setup the src dir containing the block to move
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    // Setup the dst dir whose space is totally taken by another block
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID2, mTestDir2.getCapacityBytes(),
        mTestDir2, mMetaManager, mBlockIterator);

    // session1 locks block2 first
    BlockLock lock = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID2).get();

    // Expect an exception because no eviction plan is feasible
    mThrown.expect(ResourceExhaustedRuntimeException.class);
    mThrown.expectMessage(ExceptionMessage.NO_SPACE_FOR_BLOCK_MOVE
        .getMessage(mTestDir2.toBlockStoreLocation(), BLOCK_ID1));

    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
        AllocateOptions.forMove(mTestDir2.toBlockStoreLocation()));

    // Expect createBlockMeta to succeed after unlocking this block.
    lock.close();
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
        AllocateOptions.forMove(mTestDir2.toBlockStoreLocation()));

    assertEquals(mTestDir1.getCapacityBytes(), mTestDir1.getAvailableBytes());
    assertEquals(mTestDir2.getCapacityBytes() - BLOCK_SIZE, mTestDir2.getAvailableBytes());
  }

  @Test
  public void moveBlockWithReservedSpace() throws Exception {
    Configuration.reloadProperties();
    final long RESERVE_SIZE = BLOCK_SIZE;
    init(RESERVE_SIZE);

    // Setup the src dir containing the block to move
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    // Setup the dst dir whose space is totally taken by another block
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID2,
        mTestDir2.getCapacityBytes() - BLOCK_SIZE, mTestDir2, mMetaManager, mBlockIterator);
    assertEquals(0, mTestDir2.getAvailableBytes());
    assertTrue(mTestDir2.getCapacityBytes() > mTestDir2.getCommittedBytes());

    // Expect an exception because destination is full.
    mThrown.expect(ResourceExhaustedRuntimeException.class);
    mThrown.expectMessage(ExceptionMessage.NO_SPACE_FOR_BLOCK_MOVE
        .getMessage(mTestDir2.toBlockStoreLocation(), BLOCK_ID1));

    AllocateOptions moveOptions = AllocateOptions.forMove(mTestDir2.toBlockStoreLocation())
        .setForceLocation(true)     // Stay in destination.
        .setEvictionAllowed(false); // No eviction.

    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1, moveOptions);

    // Expect createBlockMeta to succeed after setting 'useReservedSpace' flag for allocation.
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1, moveOptions.setUseReservedSpace(true));

    // Validate source is empty.
    assertEquals(mTestDir1.getCapacityBytes(),
        mTestDir1.getAvailableBytes() + mTestDir1.getReservedBytes());
    // Validate destination is full.
    assertEquals(mTestDir2.getCapacityBytes(), mTestDir2.getCommittedBytes());
  }

  /**
   * Tests that when free the space of a location, if the space of the target location is currently
   * taken by another block being locked, this freeSpace operation will fail until the lock
   * released.
   */
  @Test
  public void freeSpaceWithBlockLocked() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);

    // session1 locks a block first
    BlockLock lock = mBlockStore.pinBlock(SESSION_ID1, BLOCK_ID1).get();

    // Expect an empty eviction plan is feasible
    mThrown.expect(ResourceExhaustedRuntimeException.class);
    mThrown.expectMessage(ExceptionMessage.NO_EVICTION_PLAN_TO_FREE_SPACE.getMessage(
        mTestDir1.getCapacityBytes(), mTestDir1.toBlockStoreLocation().tierAlias()));
    mBlockStore.freeSpace(SESSION_ID1, mTestDir1.getCapacityBytes(), mTestDir1.getCapacityBytes(),
        mTestDir1.toBlockStoreLocation());

    // Expect freeSpace to succeed after unlock this block.
    lock.close();
    mBlockStore.freeSpace(SESSION_ID1, mTestDir1.getCapacityBytes(), mTestDir1.getCapacityBytes(),
        mTestDir1.toBlockStoreLocation());
    assertEquals(mTestDir1.getCapacityBytes(), mTestDir1.getAvailableBytes());
  }

  @Test
  public void getBlockWriterForNonExistingBlock() {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.TEMP_BLOCK_META_NOT_FOUND.getMessage(BLOCK_ID1));
    mBlockStore.createBlockWriter(SESSION_ID1, BLOCK_ID1);
  }

  @Test
  public void abortNonExistingBlock() {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.TEMP_BLOCK_META_NOT_FOUND.getMessage(BLOCK_ID1));
    mBlockStore.abortBlock(SESSION_ID1, BLOCK_ID1);
  }

  @Test
  public void abortBlockNotOwnedBySessionId() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.BLOCK_ID_FOR_DIFFERENT_SESSION.getMessage(TEMP_BLOCK_ID,
        SESSION_ID1, SESSION_ID2));
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.abortBlock(SESSION_ID2, TEMP_BLOCK_ID);
  }

  @Test
  public void abortCommittedBlock() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.TEMP_BLOCK_ID_COMMITTED.getMessage(TEMP_BLOCK_ID));
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.commitBlock(SESSION_ID1, TEMP_BLOCK_ID, false);
    mBlockStore.abortBlock(SESSION_ID1, TEMP_BLOCK_ID);
  }

  @Test
  public void moveNonExistingBlock() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.BLOCK_META_NOT_FOUND.getMessage(BLOCK_ID1));
    mBlockStore.moveBlock(SESSION_ID1, BLOCK_ID1,
        AllocateOptions.forMove(mTestDir1.toBlockStoreLocation()));
  }

  @Test
  public void moveTempBlock() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.BLOCK_META_NOT_FOUND.getMessage(TEMP_BLOCK_ID));
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.moveBlock(SESSION_ID1, TEMP_BLOCK_ID,
        AllocateOptions.forMove(mTestDir2.toBlockStoreLocation()));
  }

  @Test
  public void cacheSameBlockInDifferentDirs() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.ADD_EXISTING_BLOCK.getMessage(BLOCK_ID1,
        FIRST_TIER_ALIAS));
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir2, mMetaManager,
        mBlockIterator);
  }

  @Test
  public void cacheSameBlockInDifferentTiers() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.ADD_EXISTING_BLOCK.getMessage(BLOCK_ID1,
        FIRST_TIER_ALIAS));
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir3, mMetaManager,
        mBlockIterator);
  }

  @Test
  public void commitBlockTwice() throws Exception {
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    assertFalse(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    assertTrue(mBlockStore.hasTempBlockMeta(TEMP_BLOCK_ID));
    mBlockStore.commitBlock(SESSION_ID1, TEMP_BLOCK_ID , false);
    assertTrue(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    assertFalse(mBlockStore.hasTempBlockMeta(TEMP_BLOCK_ID));
    mBlockStore.commitBlock(SESSION_ID1, TEMP_BLOCK_ID, false);
    assertTrue(mBlockStore.hasBlockMeta(TEMP_BLOCK_ID));
    assertFalse(mBlockStore.hasTempBlockMeta(TEMP_BLOCK_ID));
  }

  @Test
  public void commitNonExistingBlock() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.TEMP_BLOCK_META_NOT_FOUND.getMessage(BLOCK_ID1));

    mBlockStore.commitBlock(SESSION_ID1, BLOCK_ID1, false);
  }

  @Test
  public void commitBlockNotOwnedBySessionId() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.BLOCK_ID_FOR_DIFFERENT_SESSION.getMessage(TEMP_BLOCK_ID,
        SESSION_ID1, SESSION_ID2));
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.commitBlock(SESSION_ID2, TEMP_BLOCK_ID, false);
  }

  @Test
  public void removeTempBlock() throws Exception {
    mThrown.expect(IllegalStateException.class);
    mThrown.expectMessage(ExceptionMessage.REMOVE_UNCOMMITTED_BLOCK.getMessage(TEMP_BLOCK_ID));
    TieredBlockStoreTestUtils.createTempBlock(SESSION_ID1, TEMP_BLOCK_ID, BLOCK_SIZE, mTestDir1);
    mBlockStore.removeBlock(SESSION_ID1, TEMP_BLOCK_ID);
  }

  @Test
  public void checkStorageFailedWhenInaccessible() throws Exception {
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID1, BLOCK_SIZE, mTestDir1, mMetaManager,
        mBlockIterator);
    TieredBlockStoreTestUtils.cache2(SESSION_ID1, BLOCK_ID2, BLOCK_SIZE, mTestDir2, mMetaManager,
        mBlockIterator);
    BlockStoreMeta oldMeta = mBlockStore.getBlockStoreMeta();
    FileUtils.deletePathRecursively(mTestDir2.getDirPath());
    mBlockStore.removeInaccessibleStorage();
    BlockStoreMeta meta = mBlockStore.getBlockStoreMetaFull();
    long usedByteInDir = mTestDir2.getCapacityBytes() - mTestDir2.getAvailableBytes();
    assertFalse("failed storage path should be removed",
        meta.getDirectoryPathsOnTiers().get(FIRST_TIER_ALIAS).contains(mTestDir2.getDirPath()));
    assertEquals("failed storage path quota should be deducted from store capacity",
        oldMeta.getCapacityBytes() - mTestDir2.getCapacityBytes(), meta.getCapacityBytes());
    assertEquals("failed storage path used bytes should be deducted from store used bytes",
        oldMeta.getUsedBytes() - usedByteInDir,
        meta.getUsedBytes());
    assertEquals("failed storage path quota should be deducted from tier capacity",
        oldMeta.getCapacityBytesOnTiers().get(FIRST_TIER_ALIAS) - mTestDir2.getCapacityBytes(),
        (long) meta.getCapacityBytesOnTiers().get(FIRST_TIER_ALIAS));
    assertEquals("failed storage path used bytes should be deducted from tier used bytes",
        oldMeta.getUsedBytesOnTiers().get(FIRST_TIER_ALIAS) - usedByteInDir,
        (long) meta.getUsedBytesOnTiers().get(FIRST_TIER_ALIAS));
    assertFalse("blocks in failed storage path should be removed",
        meta.getBlockList().get(FIRST_TIER_ALIAS).contains(BLOCK_ID2));
    assertTrue("blocks in working storage path should be retained",
        meta.getBlockList().get(FIRST_TIER_ALIAS).contains(BLOCK_ID1));
  }
}
