/*
 * 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.client.file.cache;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import alluxio.Constants;
import alluxio.client.file.CacheContext;
import alluxio.client.file.cache.evictor.CacheEvictor;
import alluxio.client.file.cache.evictor.FIFOCacheEvictor;
import alluxio.client.file.cache.evictor.LRUCacheEvictor;
import alluxio.client.file.cache.evictor.UnevictableCacheEvictor;
import alluxio.client.file.cache.store.LocalPageStore;
import alluxio.client.file.cache.store.LocalPageStoreDir;
import alluxio.client.file.cache.store.PageStoreDir;
import alluxio.client.file.cache.store.PageStoreOptions;
import alluxio.client.file.cache.store.PageStoreType;
import alluxio.client.quota.CacheQuota;
import alluxio.client.quota.CacheScope;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.exception.PageCorruptedException;
import alluxio.exception.PageNotFoundException;
import alluxio.exception.status.ResourceExhaustedException;
import alluxio.file.ByteArrayTargetBuffer;
import alluxio.file.NettyBufTargetBuffer;
import alluxio.file.ReadTargetBuffer;
import alluxio.network.protocol.databuffer.DataFileChannel;
import alluxio.util.CommonUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.BufferUtils;
import alluxio.util.io.FileUtils;
import alluxio.util.io.PathUtils;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.DefaultFileRegion;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.file.Paths;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * Tests for the {@link LocalCacheManager} class.
 */
public final class LocalCacheManagerTest {
  private static final int PAGE_SIZE_BYTES = Constants.KB;
  private static final int CACHE_SIZE_BYTES = 512 * Constants.KB;
  private static final PageId PAGE_ID1 = new PageId("0L", 0L);
  private static final PageId PAGE_ID2 = new PageId("1L", 1L);
  private static final byte[] PAGE1 = BufferUtils.getIncreasingByteArray(PAGE_SIZE_BYTES);
  private static final byte[] PAGE2 = BufferUtils.getIncreasingByteArray(255, PAGE_SIZE_BYTES);

  private LocalCacheManager mCacheManager;
  private InstancedConfiguration mConf = Configuration.copyGlobal();
  private PageMetaStore mPageMetaStore;
  private CacheEvictor mEvictor;
  private PageStoreOptions mPageStoreOptions;
  private byte[] mBuf = new byte[PAGE_SIZE_BYTES];

  @Rule
  public TemporaryFolder mTemp = new TemporaryFolder();
  private PageStore mPageStore;
  private LocalPageStoreDir mPageStoreDir;
  private CacheManagerOptions mCacheManagerOptions;

  @Before
  public void before() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE, PAGE_SIZE_BYTES);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(CACHE_SIZE_BYTES));
    mConf.set(PropertyKey.USER_CLIENT_CACHE_DIRS, mTemp.getRoot().getAbsolutePath());
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_WRITE_ENABLED, false);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_QUOTA_ENABLED, false);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_STORE_OVERHEAD, 0);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_STORE_TYPE, PageStoreType.LOCAL);
    // default setting in prestodb
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    // default setting in prestodb
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TIMEOUT_DURATION, "60s");
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    mPageStoreOptions = PageStoreOptions.create(mConf).get(0);
    mPageStore = PageStore.create(mPageStoreOptions);
    mEvictor = new FIFOCacheEvictor(mCacheManagerOptions.getCacheEvictorOptions());
    PageStoreDir.clear(mPageStoreOptions.getRootDir());
    mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(mPageStoreDir));
    mCacheManager = createLocalCacheManager();
  }

  @After
  public void after() throws Exception {
    mCacheManager.close();
  }

  private byte[] page(int i, int pageLen) {
    return BufferUtils.getIncreasingByteArray(i, pageLen);
  }

  private PageId pageId(long i, int pageIndex) {
    return new PageId(Long.toString(i), pageIndex);
  }

  /**
   * Creates a manager and waits until it is ready.
   */
  private LocalCacheManager createLocalCacheManager() throws Exception {
    mPageStoreOptions = PageStoreOptions.create(mConf).get(0);
    mPageStore = PageStore.create(mPageStoreOptions);
    mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
    mEvictor = new FIFOCacheEvictor(mCacheManagerOptions.getCacheEvictorOptions());
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(mPageStoreDir));
    return createLocalCacheManager(mConf, mPageMetaStore);
  }

  /**
   * Creates a manager and waits until it is ready.
   */
  private LocalCacheManager createLocalCacheManager(AlluxioConfiguration conf,
      PageMetaStore pageMetaStore) throws Exception {
    mCacheManagerOptions = CacheManagerOptions.create(conf);
    LocalCacheManager cacheManager =
        LocalCacheManager.create(mCacheManagerOptions, pageMetaStore);
    CommonUtils.waitFor("restore completed",
        () -> cacheManager.state() == CacheManager.State.READ_WRITE,
        WaitForOptions.defaults().setTimeoutMs(10000));
    return cacheManager;
  }

  @Test
  public void createNonexistentRootDirSyncRestore() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_DIRS,
        PathUtils.concatPath(mTemp.getRoot().getAbsolutePath(), UUID.randomUUID().toString()));
    mPageMetaStore =
        new DefaultPageMetaStore(PageStoreDir.createPageStoreDirs(mCacheManagerOptions));
    assertNotNull(LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore));
  }

  @Test
  public void createNonexistentRootDirAsyncRestore() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_DIRS,
        PathUtils.concatPath(mTemp.getRoot().getAbsolutePath(), UUID.randomUUID().toString()));
    mPageMetaStore =
        new DefaultPageMetaStore(PageStoreDir.createPageStoreDirs(mCacheManagerOptions));
    assertNotNull(LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore));
  }

  @Test
  public void createUnwritableRootDirSyncRestore() throws Exception {
    File root = mTemp.newFolder();
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_DIRS, root.getAbsolutePath());
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    Assume.assumeTrue(root.setWritable(false));
    try {
      mPageMetaStore =
          new DefaultPageMetaStore(PageStoreDir.createPageStoreDirs(mCacheManagerOptions));
      LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
      fail();
    } catch (Exception e) {
      // expected
    } finally {
      root.setWritable(true);
    }
  }

  @Test
  public void createUnwritableRootDirAsyncRestore() throws Exception {
    File root = mTemp.newFolder();
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_DIRS, root.getAbsolutePath());
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    Assume.assumeTrue(root.setWritable(false));
    try {
      mPageMetaStore =
          new DefaultPageMetaStore(PageStoreDir.createPageStoreDirs(mCacheManagerOptions));
      mCacheManager =
          LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
      CommonUtils.waitFor("async restore completed",
          () -> mCacheManager.state() == CacheManager.State.NOT_IN_USE,
          WaitForOptions.defaults().setTimeoutMs(10000));
    } finally {
      root.setWritable(true);
    }
  }

  @Test
  public void putNew() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void putExist() throws Exception {
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE2));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void putEvict() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mCacheManager = createLocalCacheManager();
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
  }

  @Test
  public void putSmallPages() throws Exception {
    // Cache size is only one full page, but should be able to store multiple small pages
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mCacheManager = createLocalCacheManager();
    int smallPageLen = 8;
    long numPages = mConf.getBytes(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE) / smallPageLen;
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      byte[] smallPage = page(i, smallPageLen);
      assertTrue(mCacheManager.put(id, smallPage));
    }
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      byte[] buf = new byte[smallPageLen];
      assertEquals(smallPageLen, mCacheManager.get(id, smallPageLen, buf, 0));
      assertArrayEquals(page(i, smallPageLen), buf);
    }
  }

  @Test
  public void evictSmallPageByPutSmallPage() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mCacheManager = createLocalCacheManager();
    int smallPageLen = 8;
    long numPages = mConf.getBytes(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE) / smallPageLen;
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      assertTrue(mCacheManager.put(id, page(i, smallPageLen)));
    }
    // this should trigger evicting the first page (by FIFO)
    assertTrue(mCacheManager.put(pageId(numPages, 0), page(-1, smallPageLen)));
    for (int i = 0; i < numPages; i++) {
      byte[] buf = new byte[smallPageLen];
      PageId id = pageId(i, 0);
      if (i == 0) {
        assertEquals(0, mCacheManager.get(id, smallPageLen, buf, 0));
      } else {
        assertEquals(smallPageLen, mCacheManager.get(id, smallPageLen, buf, 0));
        assertArrayEquals(page(i, smallPageLen), buf);
      }
    }
  }

  @Test
  public void evictSmallPagesByPutPigPageWithoutRetry() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mConf.set(PropertyKey.USER_CLIENT_CACHE_EVICTION_RETRIES, 0);
    mCacheManager = createLocalCacheManager();
    int smallPageLen = 8;
    long numPages = mConf.getBytes(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE) / smallPageLen;
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      assertTrue(mCacheManager.put(id, page(i, smallPageLen)));
    }
    byte[] bigPage = page(0, PAGE_SIZE_BYTES);
    PageId bigPageId = pageId(-1, 0);
    for (int i = 0; i < numPages - 1; i++) {
      assertFalse(mCacheManager.put(bigPageId, bigPage));
    }
    assertTrue(mCacheManager.put(bigPageId, bigPage));
    byte[] buf = new byte[PAGE_SIZE_BYTES];
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(bigPageId, PAGE_SIZE_BYTES, buf, 0));
    assertArrayEquals(bigPage, buf);
  }

  @Test
  public void evictSmallPagesByPutPigPageWithRetry() throws Exception {
    int smallPageLen = 8;
    int numPages = (int) (mConf.getBytes(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE) / smallPageLen);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mConf.set(PropertyKey.USER_CLIENT_CACHE_EVICTION_RETRIES, numPages);
    mCacheManager = createLocalCacheManager();
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      assertTrue(mCacheManager.put(id, page(i, smallPageLen)));
    }
    byte[] bigPage = page(0, PAGE_SIZE_BYTES);
    PageId bigPageId = pageId(-1, 0);
    assertTrue(mCacheManager.put(bigPageId, bigPage));
    byte[] buf = new byte[PAGE_SIZE_BYTES];
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(bigPageId, PAGE_SIZE_BYTES, buf, 0));
    assertArrayEquals(bigPage, buf);
  }

  @Test
  public void evictBigPagesByPutSmallPage() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    mCacheManager = createLocalCacheManager();
    PageId bigPageId = pageId(-1, 0);
    assertTrue(mCacheManager.put(bigPageId, page(0, PAGE_SIZE_BYTES)));
    int smallPageLen = 8;
    long numPages = mConf.getBytes(PropertyKey.USER_CLIENT_CACHE_PAGE_SIZE) / smallPageLen;
    byte[] smallPage = BufferUtils.getIncreasingByteArray(smallPageLen);
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      assertTrue(mCacheManager.put(id, smallPage));
    }
  }

  @Test
  public void noEvictionPolicy() throws Exception {
    mEvictor = new UnevictableCacheEvictor(mCacheManagerOptions.getCacheEvictorOptions());
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(mPageStoreDir));
    mCacheManager = createLocalCacheManager();
    long numPages = mPageStoreOptions.getCacheSize() / PAGE_SIZE_BYTES;
    for (int i = 0; i < numPages; i++) {
      PageId id = pageId(i, 0);
      assertTrue(mCacheManager.put(id, PAGE1));
    }
    assertFalse(mCacheManager.put(pageId(numPages, 0), PAGE1));
  }

  @Test
  public void putGetPartialPage() throws Exception {
    int[] sizeArray = {1, PAGE_SIZE_BYTES / 2, PAGE_SIZE_BYTES - 1};
    for (int size : sizeArray) {
      PageId pageId = new PageId("3", size);
      byte[] pageData = page(0, size);
      assertTrue(mCacheManager.put(pageId, pageData));
      byte[] buf = new byte[size];
      assertEquals(size, mCacheManager.get(pageId, size, buf, 0));
      assertArrayEquals(pageData, buf);
    }
  }

  @Test
  public void putMoreThanCacheCapacityFIFO() throws Exception {
    int cacheSize = CACHE_SIZE_BYTES / PAGE_SIZE_BYTES;
    for (int i = 0; i < 2 * cacheSize; i++) {
      PageId pageId = new PageId("3", i);
      mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES));
      if (i >= cacheSize) {
        PageId id = new PageId("3", i - cacheSize + 1);
        assertEquals(0,
            mCacheManager.get(new PageId("3", i - cacheSize), PAGE_SIZE_BYTES, mBuf, 0));
        assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(id, PAGE_SIZE_BYTES, mBuf, 0));
        assertArrayEquals(page(i - cacheSize + 1, PAGE_SIZE_BYTES), mBuf);
      }
    }
  }

  @Test
  public void putMoreThanCacheCapacityLRU() throws Exception {
    mEvictor = new LRUCacheEvictor(mCacheManagerOptions.getCacheEvictorOptions());
    mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(mPageStoreDir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    int cacheSize = CACHE_SIZE_BYTES / PAGE_SIZE_BYTES;
    //fill up the cache
    for (int i = 0; i < cacheSize; i++) {
      PageId pageId = new PageId("3", i);
      mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES));
    }
    //define page index 2,3,5 as active pages
    List<PageId> activePageIds = ImmutableList.of(2, 3, 5).stream()
        .map(pageIndex -> new PageId("3", pageIndex)).collect(
            Collectors.toList());
    //get active pages 2,3,5 to update LRU
    activePageIds.forEach(pageId -> mCacheManager.get(pageId, PAGE_SIZE_BYTES, mBuf, 0));
    //Partially fill up the cache again with new pages
    for (int i = cacheSize; i < 2 * cacheSize - 3; i++) {
      PageId pageId = new PageId("3", i);
      mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES));
    }
    //check page 2,3,5 is still in cache
    activePageIds.forEach(pageId -> {
      assertEquals(PAGE_SIZE_BYTES,
          mCacheManager.get(pageId, PAGE_SIZE_BYTES, mBuf, 0));
      assertArrayEquals(page((int) pageId.getPageIndex(), PAGE_SIZE_BYTES), mBuf);
    });
    //check the pages other than 2,3,5 got evicted
    for (int i = 0; i < cacheSize; i++) {
      PageId pageId = new PageId("3", i);
      if (!activePageIds.contains(pageId)) {
        assertEquals(0,
            mCacheManager.get(pageId, PAGE_SIZE_BYTES, mBuf, 0));
      }
    }
  }

  @Test
  public void putWithInsufficientQuota() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_QUOTA_ENABLED, true);
    mPageMetaStore = new QuotaPageMetaStore(mCacheManagerOptions.getCacheEvictorOptions(),
        ImmutableList.of(mPageStoreDir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    CacheScope scope = CacheScope.create("schema.table.partition");

    CacheContext context = CacheContext.defaults().setCacheScope(scope);
    // insufficient partition quota
    assertFalse(mCacheManager
        .put(PAGE_ID1, PAGE1, context.setCacheQuota(new CacheQuota(
            ImmutableMap.of(CacheScope.Level.PARTITION, (long) PAGE1.length - 1)))));
    // insufficient table quota
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1, context.setCacheQuota(new CacheQuota(
        ImmutableMap.of(CacheScope.Level.TABLE, (long) PAGE1.length - 1)))));
    // insufficient schema quota
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1, context.setCacheQuota(new CacheQuota(
        ImmutableMap.of(CacheScope.Level.SCHEMA, (long) PAGE1.length - 1)))));
    // insufficient global quota
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1, context.setCacheQuota(new CacheQuota(
        ImmutableMap.of(CacheScope.Level.GLOBAL, (long) PAGE1.length - 1)))));
    // without quota
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
  }

  @Test
  public void putWithQuotaEviction() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_QUOTA_ENABLED, true);
    CacheScope partitionCacheScope = CacheScope.create("schema.table.partition");
    CacheScope tableCacheScope = CacheScope.create("schema.table");
    CacheScope schemaCacheScope = CacheScope.create("schema");

    CacheScope[] quotaCacheScopes =
        {partitionCacheScope, tableCacheScope, schemaCacheScope, CacheScope.GLOBAL};
    for (CacheScope cacheScope : quotaCacheScopes) {
      mPageMetaStore = new QuotaPageMetaStore(mCacheManagerOptions.getCacheEvictorOptions(),
          ImmutableList.of(mPageStoreDir));
      PageStoreDir.clear(mPageStoreOptions.getRootDir());
      mPageStore = PageStore.create(mPageStoreOptions);
      mEvictor = CacheEvictor.create(mCacheManagerOptions.getCacheEvictorOptions());
      mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
      mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
      CacheQuota quota =
          new CacheQuota(ImmutableMap.of(cacheScope.level(),
              (long) PAGE1.length + PAGE2.length - 1));
      CacheContext context = CacheContext.defaults().setCacheScope(partitionCacheScope)
          .setCacheQuota(quota);
      assertTrue(mCacheManager.put(PAGE_ID1, PAGE1, context));
      assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
      assertTrue(mCacheManager.put(PAGE_ID2, PAGE2, context));
      assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
      assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    }
  }

  @Test
  public void putWithQuotaMoreThanCacheCapacity() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_QUOTA_ENABLED, true);
    CacheScope partitionCacheScope = CacheScope.create("schema.table.partition");
    CacheScope tableCacheScope = CacheScope.create("schema.table");
    CacheScope schemaCacheScope = CacheScope.create("schema");

    CacheScope[] quotaCacheScopes =
        {partitionCacheScope, tableCacheScope, schemaCacheScope, CacheScope.GLOBAL};
    for (CacheScope cacheScope : quotaCacheScopes) {
      mPageMetaStore = new QuotaPageMetaStore(mCacheManagerOptions.getCacheEvictorOptions(),
          ImmutableList.of(mPageStoreDir));
      mPageStoreDir.reset();
      mPageStore = PageStore.create(mPageStoreOptions);
      mEvictor = CacheEvictor.create(mCacheManagerOptions.getCacheEvictorOptions());
      mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
      mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
      CacheQuota quota = new CacheQuota(ImmutableMap.of(cacheScope.level(),
          (long) CACHE_SIZE_BYTES + 1));
      int cacheSize = CACHE_SIZE_BYTES / PAGE_SIZE_BYTES;
      for (int i = 0; i < 2 * cacheSize; i++) {
        PageId pageId = new PageId("3", i);
        CacheContext context = CacheContext.defaults().setCacheScope(partitionCacheScope)
            .setCacheQuota(quota);
        assertTrue(String.valueOf(i), mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES), context));
        if (i >= cacheSize) {
          assertEquals(
              0, mCacheManager.get(new PageId("3", i - cacheSize), PAGE_SIZE_BYTES, mBuf, 0));
          //check the subsequent page is still in cache
          assertEquals(true, mPageMetaStore.hasPage(new PageId("3", i - cacheSize + 1)));
        }
      }
    }
  }

  @Test
  public void putWithInsufficientParentQuota() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_QUOTA_ENABLED, true);
    CacheScope partitionCacheScope1 = CacheScope.create("schema.table.partition1");
    CacheScope partitionCacheScope2 = CacheScope.create("schema.table.partition2");
    CacheScope tableCacheScope = CacheScope.create("schema.table");
    CacheScope schemaCacheScope = CacheScope.create("schema");
    CacheScope[] quotaCacheScopes = {tableCacheScope, schemaCacheScope, CacheScope.GLOBAL};
    for (CacheScope cacheScope : quotaCacheScopes) {
      mPageMetaStore = new QuotaPageMetaStore(mCacheManagerOptions.getCacheEvictorOptions(),
          ImmutableList.of(mPageStoreDir));
      PageStoreDir.clear(mPageStoreOptions.getRootDir());
      mPageStore = PageStore.create(mPageStoreOptions);
      mEvictor = CacheEvictor.create(mCacheManagerOptions.getCacheEvictorOptions());
      mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
      mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
      CacheQuota quota = new CacheQuota(ImmutableMap.of(
          partitionCacheScope1.level(), (long) PAGE1.length + PAGE2.length,
          cacheScope.level(), (long) PAGE1.length + PAGE2.length - 1
      ));
      CacheContext context1 = CacheContext.defaults().setCacheScope(partitionCacheScope1)
          .setCacheQuota(quota);
      assertTrue(mCacheManager.put(PAGE_ID1, PAGE1, context1));
      assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
      CacheContext context2 = CacheContext.defaults().setCacheScope(partitionCacheScope2)
          .setCacheQuota(quota);
      assertTrue(mCacheManager.put(PAGE_ID2, PAGE2, context2));
      assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
      assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    }
  }

  @Test
  public void putLargeId() throws Exception {
    long[] fileIdArray = {Long.MAX_VALUE - 1, Long.MAX_VALUE, Long.MIN_VALUE, Long.MIN_VALUE + 1};
    long[] pageIndexArray = {Long.MAX_VALUE - 1, Long.MAX_VALUE};
    for (long fileId : fileIdArray) {
      for (long pageIndexId : pageIndexArray) {
        PageId largeId = new PageId("0", pageIndexId);
        mCacheManager.put(largeId, PAGE1);
        assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(largeId, PAGE1.length, mBuf, 0));
        assertArrayEquals(PAGE1, mBuf);
      }
    }
  }

  @Test
  public void getExist() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void getNotExist() throws Exception {
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
  }

  @Test
  public void getOffset() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    int[] offsetArray = {0, 1, PAGE_SIZE_BYTES / 2, PAGE_SIZE_BYTES - 1};
    for (int offset : offsetArray) {
      assertEquals(PAGE1.length - offset,
          mCacheManager.get(PAGE_ID1, offset, PAGE1.length - offset, mBuf, 0));
      assertEquals(ByteBuffer.wrap(PAGE1, offset, PAGE1.length - offset),
          ByteBuffer.wrap(mBuf, 0, PAGE1.length - offset));
    }
  }

  @Test
  public void getNotEnoughSpaceException() throws Exception {
    byte[] buf = new byte[PAGE1.length - 1];
    assertThrows(IllegalArgumentException.class, () ->
        mCacheManager.get(PAGE_ID1, PAGE1.length, buf, 0));
  }

  @Test
  public void deleteExist() throws Exception {
    mCacheManager.put(PAGE_ID1, PAGE1);
    mCacheManager.delete(PAGE_ID1);
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
  }

  @Test
  public void deleteNotExist() throws Exception {
    assertFalse(mCacheManager.delete(PAGE_ID1));
  }

  @Test
  public void syncRestore() throws Exception {
    mCacheManager.close();
    List<PageStoreDir> dirs = PageStoreDir.createPageStoreDirs(mCacheManagerOptions);
    PageStore pageStore = dirs.get(0).getPageStore(); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    pageStore.put(PAGE_ID1, PAGE1);
    pageStore.put(pageUuid, PAGE2);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mPageMetaStore = new DefaultPageMetaStore(dirs.subList(0, 1));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertEquals(CacheManager.State.READ_WRITE, mCacheManager.state());
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(pageUuid, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
  }

  @Test
  public void asyncRestore() throws Exception {
    mCacheManager.close();
    mPageStore = PageStore.create(mPageStoreOptions); // previous page store has been closed
    mPageStore.put(PAGE_ID1, PAGE1);
    mPageStoreDir = new LocalPageStoreDir(mPageStoreOptions, mPageStore, mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(mPageStoreDir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    mPageStore.put(PAGE_ID1, PAGE1);
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
  }

  @Test
  public void asyncRestoreReadOnly() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    mCacheManager.close();
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    SlowGetPageStoreDir slowGetPageStoreDir = new SlowGetPageStoreDir(
        mPageStoreOptions,
        new LocalPageStoreDir(mPageStoreOptions, new LocalPageStore(mPageStoreOptions), mEvictor));
    slowGetPageStoreDir.getPageStore().put(PAGE_ID1, PAGE1);
    slowGetPageStoreDir.getPageStore().put(pageUuid, PAGE2);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(slowGetPageStoreDir));
    mCacheManager = LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);

    assertEquals(CacheManager.State.READ_ONLY, mCacheManager.state());
    Thread.sleep(1000); // some buffer to restore page1
    // In READ_ONLY mode we still get previously added page
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(pageUuid, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    // In READ_ONLY mode we cannot put
    assertFalse(mCacheManager.put(PAGE_ID2, PAGE2));
    // In READ_ONLY mode we cannot delete
    assertFalse(mCacheManager.delete(PAGE_ID1));
    // stop the "fake scan"
    slowGetPageStoreDir.mScanComplete.set(true);
    CommonUtils.waitFor("async restore completed",
        () -> mCacheManager.state() == CacheManager.State.READ_WRITE,
        WaitForOptions.defaults().setTimeoutMs(10000));
    // Put get back to normal
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    assertTrue(mCacheManager.delete(PAGE_ID2));
  }

  /**
   * Invalid page file will be deleted and cache manager will start normally.
   * @throws Exception
   */
  @Test
  public void syncRestoreUnknownFile() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(pageUuid, PAGE2);
    String rootDir = mPageStoreOptions.getRootDir().toString();
    FileUtils.createFile(Paths.get(rootDir, "invalidPageFile").toString());
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
    assertEquals(CacheManager.State.READ_WRITE, mCacheManager.state());
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertEquals(PAGE2.length, mCacheManager.get(pageUuid, PAGE2.length, mBuf, 0));
  }

  /**
   * Invalid page file will be deleted and cache manager will start normally.
   * @throws Exception
   */
  @Test
  public void asyncRestoreUnknownFile() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(pageUuid, PAGE2);
    String rootDir = mPageStoreOptions.getRootDir().toString();
    FileUtils.createFile(Paths.get(rootDir, "invalidPageFile").toString());
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertEquals(PAGE2.length, mCacheManager.get(pageUuid, PAGE2.length, mBuf, 0));
  }

  @Test
  public void syncRestoreUnwritableRootDir() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(pageUuid, PAGE2);
    String rootDir = mPageStoreOptions.getRootDir().toString();
    FileUtils.deletePathRecursively(rootDir);
    File rootParent = new File(rootDir).getParentFile();
    Assume.assumeTrue(rootParent.setWritable(false));
    try {
      mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
      LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
    } catch (Exception e) {
      // expected case
    } finally {
      rootParent.setWritable(true);
    }
  }

  @Test
  public void asyncRestoreUnwritableRootDir() throws Exception {
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(pageUuid, PAGE2);
    String rootDir = mPageStoreOptions.getRootDir().toString();
    FileUtils.deletePathRecursively(rootDir);
    File rootParent = new File(rootDir).getParentFile();
    Assume.assumeTrue(rootParent.setWritable(false));
    try {
      mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
      mCacheManager = LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
      CommonUtils.waitFor("async restore completed",
          () -> mCacheManager.state() == CacheManager.State.NOT_IN_USE,
          WaitForOptions.defaults().setTimeoutMs(10000));
      assertEquals(-1, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
      assertFalse(mCacheManager.put(PAGE_ID2, PAGE2));
      assertFalse(mCacheManager.delete(PAGE_ID1));
    } finally {
      rootParent.setWritable(true);
    }
  }

  @Test
  public void syncRestoreWithMorePagesThanCapacity() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE1.length + PAGE2.length));
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, false);
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(PAGE_ID2, PAGE2);
    dir.getPageStore().put(pageUuid, BufferUtils.getIncreasingByteArray(
        PAGE1.length + PAGE2.length + 1));
    dir = PageStoreDir.createPageStoreDirs(mCacheManagerOptions).get(0);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = LocalCacheManager.create(mCacheManagerOptions, mPageMetaStore);
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    assertEquals(0, mCacheManager.get(pageUuid, PAGE2.length, mBuf, 0));
  }

  @Test
  public void asyncRestoreWithMorePagesThanCapacity() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE1.length + PAGE2.length));
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_RESTORE_ENABLED, true);
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    mCacheManager.close();
    PageStoreDir dir =
        PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
            .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(PAGE_ID2, PAGE2);
    dir.getPageStore().put(pageUuid, BufferUtils.getIncreasingByteArray(
        PAGE1.length + PAGE2.length + 1));
    dir = PageStoreDir.createPageStoreDirs(mCacheManagerOptions).get(0);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertEquals(PAGE1.length, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
    assertEquals(PAGE2.length, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    assertEquals(0, mCacheManager.get(pageUuid, PAGE2.length, mBuf, 0));
  }

  @Test
  public void ttlDeleteOldPagesWhenRestore() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TTL_THRESHOLD_SECONDS, 5);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TTL_ENABLED, true);
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    mCacheManager.close();
    PageStoreDir dir = PageStoreDir.createPageStoreDirs(mCacheManagerOptions)
                                   .get(0); // previous page store has been closed
    PageId pageUuid = new PageId(UUID.randomUUID().toString(), 0);
    dir.getPageStore().put(PAGE_ID1, PAGE1);
    dir.getPageStore().put(PAGE_ID2, PAGE2);

    dir = PageStoreDir.createPageStoreDirs(mCacheManagerOptions).get(0);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    Thread.sleep(6000);
    dir.getPageStore().put(pageUuid,
        BufferUtils.getIncreasingByteArray(PAGE1.length + PAGE2.length + 1));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertFalse(mCacheManager.hasPageUnsafe(PAGE_ID1));
    assertFalse(mCacheManager.hasPageUnsafe(PAGE_ID2));
    assertTrue(mCacheManager.hasPageUnsafe(pageUuid)); // we should have the new page
  }

  @Test
  public void asyncCache() throws Exception {
    // this must be smaller than the number of locks in the page store for the test to succeed
    final int threads = 16;
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_WRITE_ENABLED, true);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_WRITE_THREADS, threads);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_STORE_TYPE, PageStoreType.LOCAL);
    mCacheManagerOptions = CacheManagerOptions.create(mConf);
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    HangingPageStore pageStore = new HangingPageStore(pageStoreOptions);
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);

    pageStore.setPutHanging(true);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    Set<Integer> lockedPages = new HashSet<>();
    for (int i = 0; i < threads; i++) {
      PageId pageId = new PageId("5", i);
      assertTrue(mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES)));
      lockedPages.add(mCacheManager.getPageLockId(pageId));
    }
    // by setting the following line the hanging will only be stopped when the current
    // thread adds a page
    pageStore.setStopHangingThread(Thread.currentThread().getId());
    // fallback to caller's thread (the current here) when queue is full
    // find a page id that is not already locked
    int pageLockId;
    long nxtIdx = 0;
    PageId callerPageId;
    do {
      callerPageId = new PageId("0L", nxtIdx);
      pageLockId = mCacheManager.getPageLockId(callerPageId);
      nxtIdx++;
    } while (lockedPages.contains(pageLockId));
    // this page will be inserted by the current thread and not a worker thread
    assertTrue(mCacheManager.put(callerPageId, PAGE1));
    // Wait for all tasks to complete
    // one for each thread worker thread, and one on the main thread
    while (pageStore.getPuts() < threads + 1) {
      Thread.sleep(1000);
    }
    pageStore.setPutHanging(true);
    for (int i = 0; i < threads; i++) {
      PageId pageId = new PageId("6", i);
      assertTrue(mCacheManager.put(pageId, page(i, PAGE_SIZE_BYTES)));
    }
    pageStore.setPutHanging(false);
  }

  @Test
  public void asyncCacheSamePage() throws Exception {
    final int threads = 16;
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_WRITE_ENABLED, true);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_ASYNC_WRITE_THREADS, threads);
    mConf.set(PropertyKey.USER_CLIENT_CACHE_STORE_TYPE, PageStoreType.LOCAL);
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    HangingPageStore pageStore = new HangingPageStore(pageStoreOptions);
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);
    pageStore.setPutHanging(true);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1));
    pageStore.setPutHanging(false);
    while (pageStore.getPuts() < 1) {
      Thread.sleep(1000);
    }
    pageStore.setPutHanging(true);
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    pageStore.setPutHanging(false);
  }

  @Test
  public void recoverCacheFromFailedPut() throws Exception {
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    FaultyPageStore pageStore = new FaultyPageStore();
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);

    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    pageStore.setPutFaulty(true);
    // a failed put
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1));
    // no state left after previous failed put
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    // can ask to put same page again without exception
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1));
    // still no state left
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    pageStore.setPutFaulty(false);
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void failedPageStoreDeleteOnEviction() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    FaultyPageStore pageStore = new FaultyPageStore();
    LocalPageStoreDir dir = new LocalPageStoreDir(mPageStoreOptions, pageStore, mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    // first put should be ok
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    // trigger a failed eviction
    pageStore.setPutFaulty(true);
    // put operation failed because we do not retry for the IO exception from put
    assertFalse(mCacheManager.put(PAGE_ID2, PAGE2));
    //clear put faulty and set delete faulty to true
    pageStore.setPutFaulty(false);
    pageStore.setDeleteFaulty(true);
    // put operation succeeded after retry
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID2, PAGE2.length, mBuf, 0));
    assertArrayEquals(PAGE2, mBuf);
    // restore page store to function
    pageStore.setDeleteFaulty(false);
    // trigger another eviction, this should work
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    assertEquals(PAGE_SIZE_BYTES, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    assertArrayEquals(PAGE1, mBuf);
  }

  @Test
  public void putTimeout() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TIMEOUT_DURATION, "2s");
    mConf.set(PropertyKey.USER_CLIENT_CACHE_SIZE, String.valueOf(PAGE_SIZE_BYTES));
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    HangingPageStore pageStore = new HangingPageStore(pageStoreOptions);
    PageStoreDir dir = new LocalPageStoreDir(pageStoreOptions,
        new TimeBoundPageStore(pageStore, pageStoreOptions), mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    LocalCacheManager cacheManager =
        createLocalCacheManager(mConf, mPageMetaStore);
    assertTrue(cacheManager.put(PAGE_ID1, PAGE1));
    pageStore.setPutHanging(true);
    assertFalse(cacheManager.put(PAGE_ID2, PAGE2));
    pageStore.setPutHanging(false);
  }

  @Test
  public void getTimeout() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TIMEOUT_DURATION, "2s");
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    HangingPageStore pageStore = new HangingPageStore(pageStoreOptions);
    PageStoreDir dir = new LocalPageStoreDir(pageStoreOptions,
        new TimeBoundPageStore(pageStore, pageStoreOptions), mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    LocalCacheManager cacheManager =
        createLocalCacheManager(mConf, mPageMetaStore);
    assertTrue(cacheManager.put(PAGE_ID1, PAGE1));
    pageStore.setGetHanging(true);
    assertEquals(-1, cacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
    pageStore.setGetHanging(false);
  }

  @Test
  public void getFaultyReadWithNoExceptionManager() throws Exception {
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    FaultyPageStore pageStore = new FaultyPageStore();
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);

    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    NoExceptionCacheManager cacheManager =
        new NoExceptionCacheManager(createLocalCacheManager(mConf, mPageMetaStore));
    cacheManager.put(PAGE_ID1, PAGE1);
    ByteArrayTargetBuffer targetBuffer = new ByteArrayTargetBuffer(mBuf, 0);
    pageStore.setGetFaulty(true);
    assertEquals(-1, cacheManager.get(PAGE_ID1, PAGE1.length,
        targetBuffer, CacheContext.defaults()));
    assertEquals(0, targetBuffer.offset());
  }

  @Test
  public void getFaultyReadWithLocalCacheManager() throws Exception {
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    FaultyPageStore pageStore = new FaultyPageStore();
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);

    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    LocalCacheManager cacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    cacheManager.put(PAGE_ID1, PAGE1);
    ByteArrayTargetBuffer targetBuffer = new ByteArrayTargetBuffer(mBuf, 0);
    pageStore.setGetFaulty(true);
    assertEquals(-1, cacheManager.get(PAGE_ID1, PAGE1.length,
        targetBuffer, CacheContext.defaults()));
    assertEquals(0, targetBuffer.offset());
  }

  @Test
  public void getCorruptedReadWithLocalCacheManager() throws Exception {
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    FaultyPageStore pageStore = new FaultyPageStore();
    PageStoreDir dir =
        new LocalPageStoreDir(pageStoreOptions, pageStore, mEvictor);

    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    LocalCacheManager cacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    cacheManager.put(PAGE_ID1, PAGE1);
    ByteArrayTargetBuffer targetBuffer = new ByteArrayTargetBuffer(mBuf, 0);
    pageStore.setGetCorrupted(true);
    assertEquals(-1, cacheManager.get(PAGE_ID1, PAGE1.length,
        targetBuffer, CacheContext.defaults()));
    assertEquals(0, targetBuffer.offset());
  }

  @Test
  public void deleteTimeout() throws Exception {
    mConf.set(PropertyKey.USER_CLIENT_CACHE_TIMEOUT_DURATION, "2s");
    PageStoreOptions pageStoreOptions = PageStoreOptions.create(mConf).get(0);
    HangingPageStore pageStore = new HangingPageStore(pageStoreOptions);
    PageStoreDir dir = new LocalPageStoreDir(pageStoreOptions,
        new TimeBoundPageStore(pageStore, pageStoreOptions), mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    LocalCacheManager cacheManager =
        createLocalCacheManager(mConf, mPageMetaStore);
    assertTrue(cacheManager.put(PAGE_ID1, PAGE1));
    pageStore.setDeleteHanging(true);
    assertFalse(cacheManager.delete(PAGE_ID1));
    pageStore.setDeleteHanging(false);
  }

  @Test
  public void appendToPageTail() throws Exception {
    int originPageLength = 256;
    byte[] originPage = BufferUtils.getIncreasingByteArray(255, originPageLength);
    assertTrue(mCacheManager.put(PAGE_ID1, originPage));
    byte[] originPageResult = new byte[originPageLength];
    assertEquals(originPageLength,
        mCacheManager.get(PAGE_ID1, originPageLength, originPageResult, 0));
    assertArrayEquals(originPage, originPageResult);
    int appendLength = 200;
    byte[] appendContent = BufferUtils.getIncreasingByteArray(127, appendLength);
    mCacheManager.append(PAGE_ID1, originPageLength,
        appendContent, CacheContext.defaults());
    byte[] newPageResult = new byte[originPageLength + appendLength];
    assertEquals(originPageLength + appendLength,
        mCacheManager.get(PAGE_ID1, originPageLength + appendLength, newPageResult, 0));
    byte[] expectedNewPageResult = new byte[originPageLength + appendLength];
    System.arraycopy(originPage, 0, expectedNewPageResult, 0, originPageLength);
    System.arraycopy(appendContent, 0, expectedNewPageResult, originPageLength, appendLength);
    assertArrayEquals(expectedNewPageResult, newPageResult);
  }

  @Test
  public void appendToPageHead() throws Exception {
    int appendLength = 200;
    byte[] appendContent = BufferUtils.getIncreasingByteArray(127, appendLength);
    mCacheManager.append(PAGE_ID1, 0,
        appendContent, CacheContext.defaults());
    byte[] getPageResult = new byte[appendLength];
    assertEquals(appendLength,
        mCacheManager.get(PAGE_ID1, appendLength, getPageResult, 0));
    assertArrayEquals(appendContent, getPageResult);
  }

  @Test
  public void noSpaceLeftPageStorePut() throws Exception {
    LocalPageStore pageStore = new LocalPageStore(
        PageStoreOptions.create(mConf).get(0)) {
      private long mFreeBytes = PAGE_SIZE_BYTES;

      @Override
      public void delete(PageId pageId) throws IOException, PageNotFoundException {
        mFreeBytes += PAGE_SIZE_BYTES;
        super.delete(pageId);
      }

      @Override
      public void put(PageId pageId, ByteBuffer page, boolean isTempory) throws IOException {
        if (mFreeBytes < page.remaining()) {
          throw new ResourceExhaustedException("No space left on device");
        }
        mFreeBytes -= page.remaining();
        super.put(pageId, page, isTempory);
      }
    };
    PageStoreDir dir =
        new LocalPageStoreDir(PageStoreOptions.create(mConf).get(0),
            pageStore, mEvictor);
    mPageMetaStore = new DefaultPageMetaStore(ImmutableList.of(dir));
    mCacheManager = createLocalCacheManager(mConf, mPageMetaStore);
    assertTrue(mCacheManager.put(PAGE_ID1, PAGE1));
    // trigger evicting PAGE1
    assertTrue(mCacheManager.put(PAGE_ID2, PAGE2));
    assertEquals(0, mCacheManager.get(PAGE_ID1, PAGE1.length, mBuf, 0));
  }

  @Test
  public void highStorageOverheadPut() throws Exception {
    // a store that is so inefficient to store any data
    double highOverhead = CACHE_SIZE_BYTES / PAGE_SIZE_BYTES + 0.1;
    mConf.set(PropertyKey.USER_CLIENT_CACHE_STORE_OVERHEAD, highOverhead);
    mCacheManager = createLocalCacheManager();
    assertFalse(mCacheManager.put(PAGE_ID1, PAGE1));
  }

  @Test
  public void listPageIds() throws Exception {
    mCacheManager = createLocalCacheManager();
    assertEquals(0,
        mCacheManager.getCachedPageIdsByFileId(PAGE_ID1.getFileId(), 64 * PAGE_SIZE_BYTES).size());
    mCacheManager.put(PAGE_ID1, PAGE1);
    assertEquals(PAGE_ID1,
        mCacheManager.getCachedPageIdsByFileId(PAGE_ID1.getFileId(), 64 * PAGE_SIZE_BYTES).get(0));
    PageId pageId5 = new PageId(PAGE_ID1.getFileId(), 5);
    mCacheManager.put(pageId5, PAGE1);
    assertEquals(PAGE_ID1,
        mCacheManager.getCachedPageIdsByFileId(PAGE_ID1.getFileId(), 64 * PAGE_SIZE_BYTES).get(0));
    assertEquals(pageId5,
        mCacheManager.getCachedPageIdsByFileId(PAGE_ID1.getFileId(), 64 * PAGE_SIZE_BYTES).get(1));
    //Store a page smaller than full size
    PageId pageId6 = new PageId(PAGE_ID1.getFileId(), 6);
    mCacheManager.put(pageId6, BufferUtils.getIncreasingByteArray(135));
    assertEquals(pageId6,
        mCacheManager.getCachedPageIdsByFileId(PAGE_ID1.getFileId(),
            6 * PAGE_SIZE_BYTES + 135).get(2));
    //Store a file smaller than one page
    PageId smallFilePageId = new PageId("small_file", 0);
    mCacheManager.put(smallFilePageId, BufferUtils.getIncreasingByteArray(135));
    assertEquals(smallFilePageId,
        mCacheManager.getCachedPageIdsByFileId(smallFilePageId.getFileId(),
            135).get(0));
    //Store a zero length file
    PageId zeroLenFilePageId = new PageId("zero_len_file", 0);
    mCacheManager.put(zeroLenFilePageId, BufferUtils.getIncreasingByteArray(0));
    assertEquals(zeroLenFilePageId,
        mCacheManager.getCachedPageIdsByFileId(zeroLenFilePageId.getFileId(),
            0).get(0));
  }

  @Test
  public void getDataFileChannel() throws Exception {
    mCacheManager = createLocalCacheManager();
    mCacheManager.put(PAGE_ID1, PAGE1);
    CacheContext cacheContext = CacheContext.defaults();
    Optional<DataFileChannel> dataFileChannel = mCacheManager.getDataFileChannel(PAGE_ID1,
        0, PAGE1.length, cacheContext);
    assertNotNull(dataFileChannel);
    assertEquals(dataFileChannel.isPresent(), true);
    assertEquals(dataFileChannel.get().getNettyOutput() instanceof DefaultFileRegion, true);
    DefaultFileRegion defaultFileRegion =
        (DefaultFileRegion) dataFileChannel.get().getNettyOutput();
    ByteBuf buf = Unpooled.buffer(PAGE1.length);
    NettyBufTargetBuffer targetBuffer = new NettyBufTargetBuffer(buf);
    long bytesTransferred = defaultFileRegion.transferTo(targetBuffer.byteChannel(), 0);
    assertEquals(bytesTransferred, PAGE1.length);

    byte[] bytes = new byte[PAGE1.length];
    buf.readBytes(bytes);
    assertArrayEquals(PAGE1, bytes);
  }

  /**
   * A PageStore where put can throw IOException on put or delete.
   */
  private class FaultyPageStore extends LocalPageStore {
    public FaultyPageStore() {
      super(PageStoreOptions.create(mConf).get(0));
    }

    private AtomicBoolean mPutFaulty = new AtomicBoolean(false);
    private AtomicBoolean mDeleteFaulty = new AtomicBoolean(false);
    private AtomicBoolean mGetFaulty = new AtomicBoolean(false);

    private AtomicBoolean mGetCorrupted = new AtomicBoolean(false);

    @Override
    public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer target,
        boolean isTemporary) throws IOException, PageNotFoundException {
      if (mGetFaulty.get()) {
        target.offset(target.offset() + 100);
        throw new IOException("Page read fault");
      }
      if (mGetCorrupted.get()) {
        target.offset(target.offset() + 100);
        throw new PageCorruptedException("page corrupted");
      }
      return super.get(pageId, pageOffset, bytesToRead, target, isTemporary);
    }

    @Override
    public void put(PageId pageId, ByteBuffer page, boolean isTemporary) throws IOException {
      if (mPutFaulty.get()) {
        throw new IOException("Not found");
      }
      super.put(pageId, page, isTemporary);
    }

    @Override
    public void delete(PageId pageId) throws IOException, PageNotFoundException {
      if (mDeleteFaulty.get()) {
        throw new IOException("Not found");
      }
      super.delete(pageId);
    }

    void setPutFaulty(boolean faulty) {
      mPutFaulty.set(faulty);
    }

    void setDeleteFaulty(boolean faulty) {
      mDeleteFaulty.set(faulty);
    }

    void setGetFaulty(boolean faulty) {
      mGetFaulty.set(faulty);
    }

    void setGetCorrupted(boolean faulty) {
      mGetCorrupted.set(faulty);
    }
  }

  /**
   * A PageStore with slow scan.
   */
  private class SlowGetPageStoreDir extends LocalPageStoreDir {

    private final PageStoreDir mUnderPageStoreDir;

    private AtomicInteger mPageId = new AtomicInteger(100);
    private AtomicBoolean mScanComplete = new AtomicBoolean(false);

    public SlowGetPageStoreDir(PageStoreOptions pageStoreOptions,
        PageStoreDir pageStoreDir) {
      super(pageStoreOptions, pageStoreDir.getPageStore(), mEvictor);
      mUnderPageStoreDir = pageStoreDir;
    }

    @Override
    public void scanPages(Consumer<Optional<PageInfo>> pageInfoConsumer) throws IOException {
      mUnderPageStoreDir.scanPages(pageInfoConsumer);
      while (!mScanComplete.get()) {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
        PageId id = new PageId(String.valueOf(mPageId.getAndIncrement()), 0L);
        pageInfoConsumer.accept(Optional.of(new PageInfo(id, 1L, this)));
      }
    }
  }
}
