public static List<Path> listTaskTemporaryPaths(
            FileSystem fs, Path basePath, BiPredicate<Integer, Integer> taskAttemptFilter)
            throws Exception {
        List<Path> taskTmpPaths = new ArrayList<>();

        if (fs.exists(basePath)) {
            for (FileStatus taskStatus : fs.listStatus(basePath)) {
                final String taskDirName = taskStatus.getPath().getName();
                final Matcher matcher = TASK_DIR_PATTERN.matcher(taskDirName);
                if (matcher.matches()) {
                    final int subtaskIndex = Integer.parseInt(matcher.group(1));
                    final int attemptNumber = Integer.parseInt(matcher.group(2));
                    if (taskAttemptFilter.test(subtaskIndex, attemptNumber)) {
                        taskTmpPaths.add(taskStatus.getPath());
                    }
                }
            }
        } else {
            LOG.warn(
                    "The path {} doesn't exist. Maybe no data is generated in the path and the path is not created.",
                    basePath);
        }
        return taskTmpPaths;
    }
@Test
    void testListTaskTemporaryPaths() throws Exception {
        // only accept task-0-attempt-1
        final BiPredicate<Integer, Integer> taskAttemptFilter =
                (subtaskIndex, attemptNumber) -> subtaskIndex == 0 && attemptNumber == 1;

        final FileSystem fs = FileSystem.get(tmpPath.toUri());
        fs.mkdirs(new Path(tmpPath.toUri() + "/task-0-attempt-0")); // invalid attempt number
        fs.mkdirs(new Path(tmpPath.toUri() + "/task-0-attempt-1")); // valid
        fs.mkdirs(new Path(tmpPath.toUri() + "/task-1-attempt-0")); // invalid subtask index
        fs.mkdirs(new Path(tmpPath.toUri() + "/.task-0-attempt-1")); // invisible dir
        fs.mkdirs(new Path(tmpPath.toUri() + "/_SUCCESS")); // not a task dir

        final List<Path> taskTmpPaths =
                PartitionTempFileManager.listTaskTemporaryPaths(
                        fs, new Path(tmpPath.toUri()), taskAttemptFilter);
        final List<String> taskDirs =
                taskTmpPaths.stream().map(Path::getName).collect(Collectors.toList());
        assertThat(taskDirs).hasSize(1).containsExactly("task-0-attempt-1");
    }