@Override
  public Optional<DataFileChannel> getDataFileChannel(
      PageId pageId, int pageOffset, int bytesToRead, CacheContext cacheContext)
      throws PageNotFoundException {
    Preconditions.checkArgument(pageOffset <= mOptions.getPageSize(),
        "Read exceeds page boundary: offset=%s size=%s",
        pageOffset, mOptions.getPageSize());
    LOG.debug("get({},pageOffset={}) enters", pageId, pageOffset);
    if (mState.get() == NOT_IN_USE) {
      Metrics.GET_NOT_READY_ERRORS.inc();
      Metrics.GET_ERRORS.inc();
      throw new PageNotFoundException(String.format("Page %s could not be found", pageId));
    }
    ReadWriteLock pageLock = getPageLock(pageId);
    long startTime = System.nanoTime();
    try (LockResource r = new LockResource(pageLock.readLock())) {
      PageInfo pageInfo;
      try (LockResource r2 = new LockResource(mPageMetaStore.getLock().readLock())) {
        pageInfo = mPageMetaStore.getPageInfo(pageId); //check if page exists and refresh LRU items
      } catch (PageNotFoundException e) {
        LOG.debug("getDataChannel({},pageOffset={}) fails due to page not found in metastore",
            pageId, pageOffset);
        return Optional.empty();
      }

      try {
        DataFileChannel dataFileChannel = pageInfo.getLocalCacheDir().getPageStore()
            .getDataFileChannel(pageInfo.getPageId(), pageOffset, bytesToRead,
                cacheContext.isTemporary());
        MultiDimensionalMetricsSystem.CACHED_DATA_READ.inc(bytesToRead);
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_HIT_REQUESTS.getName()).inc();
        MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getName()).mark(bytesToRead);
        cacheContext.incrementCounter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getMetricName(), BYTE,
            bytesToRead);
        LOG.debug("getDataChannel({},pageOffset={}) exits", pageId, pageOffset);
        return Optional.of(dataFileChannel);
      } catch (PageNotFoundException e) {
        LOG.debug("getDataChannel({},pageOffset={}) fails due to page file not found",
            pageId, pageOffset);
        Metrics.GET_ERRORS.inc();
        Metrics.GET_STORE_READ_ERRORS.inc();
        // something is wrong to read this page, let's remove it from meta store
        try (LockResource r2 = new LockResource(mPageMetaStore.getLock().writeLock())) {
          mPageMetaStore.removePage(pageId);
          return Optional.empty();
        } catch (PageNotFoundException ex) {
          // best effort to remove this page from meta store and ignore the exception
          Metrics.CLEANUP_GET_ERRORS.inc();
          return Optional.empty();
        }
      }
    } finally {
      cacheContext.incrementCounter(
          MetricKey.CLIENT_CACHE_PAGE_READ_CACHE_TIME_NS.getMetricName(), NANO,
          System.nanoTime() - startTime);
    }
  }
@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);
  }