public static List<String> findInstantFilesToRemove(
      String instantToRepair, List<String> baseAndLogFilesFromFs,
      HoodieActiveTimeline activeTimeline, HoodieArchivedTimeline archivedTimeline) {
    // Skips the instant if it is requested or inflight in active timeline
    if (!activeTimeline.filter(instant -> instant.getTimestamp().equals(instantToRepair)
        && !instant.isCompleted()).empty()) {
      return Collections.emptyList();
    }

    try {
      boolean doesInstantExist = false;
      Option<Set<String>> filesFromTimeline = Option.empty();
      Option<HoodieInstant> instantOption = activeTimeline.filterCompletedInstants().filter(
          instant -> instant.getTimestamp().equals(instantToRepair)).firstInstant();
      if (instantOption.isPresent()) {
        // Completed instant in active timeline
        doesInstantExist = true;
        filesFromTimeline = RepairUtils.getBaseAndLogFilePathsFromTimeline(
            activeTimeline, instantOption.get());
      } else {
        instantOption = archivedTimeline.filterCompletedInstants().filter(
            instant -> instant.getTimestamp().equals(instantToRepair)).firstInstant();
        if (instantOption.isPresent()) {
          // Completed instant in archived timeline
          doesInstantExist = true;
          filesFromTimeline = RepairUtils.getBaseAndLogFilePathsFromTimeline(
              archivedTimeline, instantOption.get());
        }
      }

      if (doesInstantExist) {
        if (!filesFromTimeline.isPresent() || filesFromTimeline.get().isEmpty()) {
          // Skips if no instant details
          return Collections.emptyList();
        }
        // Excludes committed base and log files from timeline
        Set<String> filesToRemove = new HashSet<>(baseAndLogFilesFromFs);
        filesToRemove.removeAll(filesFromTimeline.get());
        return new ArrayList<>(filesToRemove);
      } else {
        // The instant does not exist in the whole timeline (neither completed nor requested/inflight),
        // this means the files from this instant are dangling, which should be removed
        return baseAndLogFilesFromFs;
      }
    } catch (IOException e) {
      // In case of failure, does not remove any files for the instant
      return Collections.emptyList();
    }
  }
@Test
  public void testFindInstantFilesToRemove() throws IOException {
    setupTimelineInFS();
    HoodieInstant existingInstant = new HoodieInstant(
        HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
    Map<String, List<Pair<String, String>>> partitionToFileIdAndNameMap =
        instantInfoMap.get(existingInstant.getTimestamp());
    List<String> fileListFromFs = partitionToFileIdAndNameMap.entrySet().stream()
        .flatMap(entry ->
            entry.getValue().stream()
                .map(fileInfo -> new StoragePath(entry.getKey(), fileInfo.getValue()).toString())
                .collect(Collectors.toList())
                .stream()
        ).collect(Collectors.toList());
    String danglingFilePath = new StoragePath("2022/01/02",
        getBaseFilename(existingInstant.getTimestamp(), UUID.randomUUID().toString())).toString();
    fileListFromFs.add(danglingFilePath);
    // Existing instant
    assertEquals(CollectionUtils.createImmutableList(danglingFilePath),
        RepairUtils.findInstantFilesToRemove(
            existingInstant.getTimestamp(), fileListFromFs,
            metaClient.getActiveTimeline(), metaClient.getArchivedTimeline()));
    // Non-existing instant
    assertEquals(fileListFromFs,
        RepairUtils.findInstantFilesToRemove(
            "004", fileListFromFs,
            metaClient.getActiveTimeline(), metaClient.getArchivedTimeline()));
  }