@Override
  public boolean put(PageId pageId, ByteBuffer page, CacheContext cacheContext) {
    LOG.debug("put({},{} bytes) enters", pageId, page.remaining());
    if (mState.get() != READ_WRITE) {
      Metrics.PUT_NOT_READY_ERRORS.inc();
      Metrics.PUT_ERRORS.inc();
      return false;
    }
    int originPosition = page.position();
    if (!mOptions.isAsyncWriteEnabled()) {
      boolean ok = putInternal(pageId, page, cacheContext);
      LOG.debug("put({},{} bytes) exits: {}", pageId, page.position() - originPosition, ok);
      if (!ok) {
        Metrics.PUT_ERRORS.inc();
      }
      return ok;
    }

    if (!mPendingRequests.add(pageId)) { // already queued
      return false;
    }
    try {
      mAsyncCacheExecutor.get().submit(() -> {
        try {
          boolean ok = putInternal(pageId, page, cacheContext);
          if (!ok) {
            Metrics.PUT_ERRORS.inc();
          }
        } finally {
          mPendingRequests.remove(pageId);
        }
      });
    } catch (RejectedExecutionException e) { // queue is full, skip
      // RejectedExecutionException may be thrown in extreme cases when the
      // highly concurrent caching workloads. In these cases, return false
      mPendingRequests.remove(pageId);
      Metrics.PUT_ASYNC_REJECTION_ERRORS.inc();
      Metrics.PUT_ERRORS.inc();
      LOG.debug("put({},{} bytes) fails due to full queue", pageId,
          page.position() - originPosition);
      return false;
    }
    LOG.debug("put({},{} bytes) exits with async write", pageId, page.position() - originPosition);
    return true;
  }
@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));
  }