@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 testLoadFromReader() throws IOException {
    String ufsPath = "testLoadRemote";
    mWorker.loadDataFromRemote(ufsPath, 0, 10, new TestDataReader(100), (int) mPageSize);
    byte[] buffer = new byte[10];
    String fileId = new AlluxioURI(ufsPath).hash();
    List<PageId> cachedPages = mCacheManager.getCachedPageIdsByFileId(fileId, 10);
    assertEquals(1, cachedPages.size());
    mCacheManager.get(new PageId(fileId, 0), 10, buffer, 0);
    assertTrue(BufferUtils.equalIncreasingByteArray(0, 10, buffer));
  }