code: @Override
    public Collection<CopyableFile> filter(FileSystem sourceFs, FileSystem targetFs,
        Collection<CopyableFile> copyableFiles) {
      Iterator<CopyableFile> cfIterator = copyableFiles.iterator();
  
      ImmutableList.Builder<CopyableFile> filtered = ImmutableList.builder();
  
      while (cfIterator.hasNext()) {
        CopyableFile cf = cfIterator.next();
        Path readyFilePath = PathUtils.addExtension(cf.getOrigin().getPath(), READY_EXTENSION);
        try {
          if (sourceFs.exists(readyFilePath)) {
            filtered.add(cf);
          } else {
            log.info(String.format("Removing %s as the .ready file is not found", cf.getOrigin().getPath()));
          }
        } catch (IOException e) {
          log.warn(String.format("Removing %s as the .ready file can not be read. Exception %s",
              cf.getOrigin().getPath(), e.getMessage()));
        }
      }
      return filtered.build();
    }
docs: For every {@link CopyableFile} in <code>copyableFiles</code> checks if a {@link CopyableFile#getOrigin()#getPath()}
  + .ready files is present on <code>sourceFs</code> {@inheritDoc}
  
  @see org.apache.gobblin.data.management.copy.CopyableFileFilter#filter(org.apache.hadoop.fs.FileSystem,
  org.apache.hadoop.fs.FileSystem, java.util.Collection)