@VisibleForTesting
  public void loadDataFromRemote(String filePath, long offset, long lengthToLoad,
      PositionReader reader, int chunkSize) throws IOException {
    ByteBuffer buf = ByteBuffer.allocateDirect(chunkSize);
    String fileId = new AlluxioURI(filePath).hash();
    while (lengthToLoad > 0) {
      long currentPageIndex = offset / mPageSize;
      PageId pageId = new PageId(fileId.toString(), currentPageIndex);
      int lengthToRead = (int) Math.min(chunkSize, lengthToLoad);
      int lengthRead = reader.read(offset, buf, lengthToRead);
      if (lengthRead != lengthToRead) {
        throw new FailedPreconditionRuntimeException(
            "Read " + lengthRead + " bytes, expected to read " + lengthToRead + " bytes");
      }
      buf.flip();
      mCacheManager.put(pageId, buf);
      offset += lengthRead;
      lengthToLoad -= lengthRead;
      buf.clear();
    }
  }
@Test
  public void testLoadBlockFromReader() throws IOException {
    String ufsPath = "testLoadBlockRemote";
    long offset = mPageSize;
    int numPages = 3;
    long lengthToLoad = numPages * mPageSize + 5;
    mWorker.loadDataFromRemote(ufsPath, offset, lengthToLoad,
        new TestDataReader((int) (5 * mPageSize)), (int) mPageSize);
    String fileId = new AlluxioURI(ufsPath).hash();
    List<PageId> cachedPages = mCacheManager.getCachedPageIdsByFileId(fileId, 5 * mPageSize);
    assertEquals(4, cachedPages.size());
    for (int i = 1; i < 4; i++) {
      byte[] buffer = new byte[(int) mPageSize];
      mCacheManager.get(new PageId(fileId, i), (int) mPageSize, buffer, 0);
      assertTrue(BufferUtils.equalIncreasingByteArray((int) (offset + (i - 1) * mPageSize),
          (int) mPageSize, buffer));
    }
    // test last page with 5 bytes
    byte[] buffer = new byte[(int) 5];
    mCacheManager.get(new PageId(fileId, 4), 5, buffer, 0);
    assertTrue(BufferUtils.equalIncreasingByteArray((int) (offset + 3 * mPageSize), 5, buffer));
  }