@Override
  public synchronized List<PersistFile> pollFilesToPersist(long workerId)
      throws InvalidPathException, AccessControlException {
    List<PersistFile> filesToPersist = new ArrayList<>();
    List<Long> fileIdsToPersist = new ArrayList<>();

    if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) {
      return filesToPersist;
    }

    Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId);
    try {
      for (long fileId : scheduledFiles) {
        try {
          FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId);
          if (fileInfo.isCompleted()) {
            fileIdsToPersist.add(fileId);
            List<Long> blockIds = new ArrayList<>();
            for (FileBlockInfo fileBlockInfo : mFileSystemMasterView
                .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) {
              blockIds.add(fileBlockInfo.getBlockInfo().getBlockId());
            }

            filesToPersist.add(new PersistFile(fileId, blockIds));
          }
        } catch (FileDoesNotExistException e) {
          LOG.warn("FileId {} does not exist, ignore persistence it", fileId);
        }
      }
    } catch (UnavailableException e) {
      return filesToPersist;
    }

    mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist);
    return filesToPersist;
  }
@Test
  public void persistenceFileWithBlocksOnMultipleWorkers() throws Exception {
    DefaultAsyncPersistHandler handler =
        new DefaultAsyncPersistHandler(new FileSystemMasterView(mFileSystemMaster));
    AlluxioURI path = new AlluxioURI("/test");
    List<FileBlockInfo> blockInfoList = new ArrayList<>();
    BlockLocation location1 = new BlockLocation().setWorkerId(1);
    blockInfoList.add(new FileBlockInfo()
        .setBlockInfo(new BlockInfo().setLocations(Lists.newArrayList(location1))));
    BlockLocation location2 = new BlockLocation().setWorkerId(2);
    blockInfoList.add(new FileBlockInfo()
        .setBlockInfo(new BlockInfo().setLocations(Lists.newArrayList(location2))));
    long fileId = 2;
    when(mFileSystemMaster.getFileId(path)).thenReturn(fileId);
    when(mFileSystemMaster.getFileInfo(fileId))
        .thenReturn(new FileInfo().setLength(1).setCompleted(true));
    when(mFileSystemMaster.getFileBlockInfoList(path)).thenReturn(blockInfoList);

    // no persist scheduled on any worker
    assertEquals(0, handler.pollFilesToPersist(1).size());
    assertEquals(0, handler.pollFilesToPersist(2).size());
  }