code: @Override
    public synchronized List<PersistFile> pollFilesToPersist(long workerId)
        throws FileDoesNotExistException, 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;
    }
docs: Polls the files to send to the given worker for persistence. It also removes files from the
  worker entry in {@link #mWorkerToAsyncPersistFiles}.
  
  @param workerId the worker id
  @return the list of files
  @throws FileDoesNotExistException if the file does not exist
  @throws InvalidPathException if the path is invalid
  @throws AccessControlException if permission checking fails