@Override
  public ListenableFuture<LoadFileResponse> load(List<LoadSubTask> subTasks, boolean skipIfExists,
      UfsReadOptions options) throws AccessControlException, IOException {
    List<ListenableFuture<Void>> futures = new ArrayList<>();
    List<LoadFailure> errors = Collections.synchronizedList(new ArrayList<>());
    AtomicInteger numSkipped = new AtomicInteger();
    AtomicLong skippedLength = new AtomicLong();
    for (LoadSubTask task : subTasks) {
      if (task.hasLoadMetadataSubtask()) {
        UfsStatus status = UfsStatus.fromProto(task.getLoadMetadataSubtask().getUfsStatus());
        loadMetadata(status, errors);
      }
      if (task.hasLoadDataSubtask()) {
        LoadDataSubTask subtask = task.getLoadDataSubtask();
        if (subtask.getLength() <= 0) {
          continue;
        }
        boolean countAsSkipped = skipIfExists && isAllPageCached(subtask.getUfsStatus(),
            subtask.getOffsetInFile(), subtask.getLength());
        if (countAsSkipped) {
          numSkipped.incrementAndGet();
          skippedLength.addAndGet(subtask.getLength());
          continue;
        }
        try {
          ListenableFuture<Void> loadFuture = submitLoadDataSubTask(subtask, options, errors);
          futures.add(loadFuture);
        } catch (RejectedExecutionException ex) {
          LOG.warn("Load task overloaded.");
          errors.add(LoadFailure.newBuilder().setSubtask(
                                    LoadSubTask.newBuilder().setLoadDataSubtask(subtask).build())
                                .setCode(Status.RESOURCE_EXHAUSTED.getCode().value())
                                .setRetryable(true).setMessage(ex.getMessage()).build());
        }
      }
    }
    return Futures.whenAllComplete(futures).call(
        () -> LoadFileResponse.newBuilder().addAllFailures(errors)
            .setBytesSkipped(skippedLength.get()).setNumSkipped(numSkipped.get())
            // Status is a required field, put it as a placeholder
            .setStatus(TaskStatus.SUCCESS).build(),
        GrpcExecutors.READER_EXECUTOR);
  }
@Test
  public void testLoad() throws Exception {
    int numPages = 10;
    long length = mPageSize * numPages;
    String ufsPath = mTestFolder.newFile("test").getAbsolutePath();
    byte[] buffer = BufferUtils.getIncreasingByteArray((int) length);
    BufferUtils.writeBufferToFile(ufsPath, buffer);
    loadFileData(ufsPath);
    List<PageId> cachedPages =
        mCacheManager.getCachedPageIdsByFileId(new AlluxioURI(ufsPath).hash(), length);
    assertEquals(numPages, cachedPages.size());
    int start = 0;
    for (PageId pageId : cachedPages) {
      byte[] buff = new byte[(int) mPageSize];
      mCacheManager.get(pageId, (int) mPageSize, buff, 0);
      assertTrue(BufferUtils.equalIncreasingByteArray(start, (int) mPageSize, buff));
      start += mPageSize;
    }
    assertTrue(mWorker.getMetaManager().getFromMetaStore(ufsPath).isPresent());
  }