@Override
  // TODO(yimin) integrate this method with load() method
  public void cacheData(String ufsPath, long length, long pos, boolean isAsync)
      throws IOException {
    List<CompletableFuture<Void>> futures = new ArrayList<>();
    // TODO(yimin) To implement the sync data caching.
    alluxio.grpc.FileInfo fi = getGrpcFileInfo(ufsPath, -1);
    String fileId = new AlluxioURI(ufsPath).hash();
    for (long i = pos / mPageSize;
         i <= Math.min(pos + length, fi.getLength()) / mPageSize; ++i) {
      PageId pageId = new PageId(fileId, i);
      // TODO(yimin) As an optimization, data does not need to load on a page basis.
      // Can implement a bulk load mechanism and load a couple of pages at the same time,
      // to improve the performance.
      if (mCacheManager.hasPageUnsafe(pageId)) {
        continue;
      }
      long loadPos = i * mPageSize;
      long loadLength = Math.min(mPageSize, fi.getLength() - loadPos);
      if (loadLength == 0) {
        continue;
      }
      if (!mLoadingPages.addIfAbsent(pageId)) {
        continue;
      }

      futures.add(CompletableFuture.runAsync(() -> {
        try {
          if (mCacheManager.hasPageUnsafe(pageId)) {
            return;
          }
          LOG.debug("Preloading {} pos: {} length: {} started", ufsPath, loadPos, loadLength);
          loadPages(ufsPath, Collections.singletonList(pageId), fi.getLength());
          LOG.debug("Preloading {} pos: {} length: {} finished", ufsPath, loadPos, loadLength);
        } catch (Exception e) {
          LOG.info("Preloading failed for {} page: {}", ufsPath, pageId, e);
        } finally {
          mLoadingPages.remove(pageId);
        }
      }, mCacheDataExecutor));
      if (!isAsync) {
        try {
          CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      }
    }
  }
@Test
  public void testCacheDataNotPageAligned() throws Exception {
    int numPages = 10;
    long length = mPageSize * numPages - 1;
    String ufsPath = mTestFolder.newFile("test").getAbsolutePath();
    byte[] buffer = BufferUtils.getIncreasingByteArray((int) length);
    BufferUtils.writeBufferToFile(ufsPath, buffer);

    mWorker.cacheData(ufsPath, length, 0, false);
    List<PageId> cachedPages =
        mCacheManager.getCachedPageIdsByFileId(new AlluxioURI(ufsPath).hash(), length);
    assertEquals(numPages, cachedPages.size());
    int start = 0;
    for (PageId pageId : cachedPages) {
      long size = numPages == pageId.getPageIndex() + 1 ? length % mPageSize : mPageSize;
      byte[] buff = new byte[(int) size];
      mCacheManager.get(pageId, (int) size, buff, 0);
      assertTrue(BufferUtils.equalIncreasingByteArray(start, (int) size, buff));
      start += mPageSize;
    }
  }