@ShellMethod(key = "repair corrupted clean files", value = "repair corrupted clean files")
  public void removeCorruptedPendingCleanAction() {

    HoodieTableMetaClient client = HoodieCLI.getTableMetaClient();
    HoodieTimeline cleanerTimeline = HoodieCLI.getTableMetaClient().getActiveTimeline().getCleanerTimeline();
    LOG.info("Inspecting pending clean metadata in timeline for corrupted files");
    cleanerTimeline.filterInflightsAndRequested().getInstants().forEach(instant -> {
      try {
        CleanerUtils.getCleanerPlan(client, instant);
      } catch (AvroRuntimeException e) {
        LOG.warn("Corruption found. Trying to remove corrupted clean instant file: " + instant);
        HoodieActiveTimeline.deleteInstantFile(client.getStorage(), client.getMetaPath(),
            instant);
      } catch (IOException ioe) {
        if (ioe.getMessage().contains("Not an Avro data file")) {
          LOG.warn("Corruption found. Trying to remove corrupted clean instant file: " + instant);
          HoodieActiveTimeline.deleteInstantFile(client.getStorage(), client.getMetaPath(),
              instant);
        } else {
          throw new HoodieIOException(ioe.getMessage(), ioe);
        }
      }
    });
  }
@Test
  public void testRemoveCorruptedPendingCleanAction() throws IOException {
    HoodieCLI.conf = storageConf();

    StorageConfiguration<?> conf = HoodieCLI.conf;

    HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();

    // Create four requested files
    for (int i = 100; i < 104; i++) {
      String timestamp = String.valueOf(i);
      // Write corrupted requested Clean File
      HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath, timestamp, conf);
    }

    // reload meta client
    metaClient = HoodieTableMetaClient.reload(metaClient);
    // first, there are four instants
    assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());

    Object result = shell.evaluate(() -> "repair corrupted clean files");
    assertTrue(ShellEvaluationResultUtil.isSuccess(result));

    // reload meta client
    metaClient = HoodieTableMetaClient.reload(metaClient);
    assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
  }