@Test public void testRemoveCorruptedPendingCleanAction() throws IOException { HoodieCLI.conf = jsc.hadoopConfiguration(); Configuration conf = HoodieCLI.conf; metaClient = HoodieCLI.getTableMetaClient(); for (int i = 100; i < 104; i++) { String timestamp = String.valueOf(i); HoodieTestCommitMetadataGenerator.createEmptyCleanRequestedFile(tablePath, timestamp, conf); } metaClient = HoodieTableMetaClient.reload(metaClient); assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count()); CommandResult cr = getShell().executeCommand("repair corrupted clean files"); assertTrue(cr.isSuccess()); metaClient = HoodieTableMetaClient.reload(metaClient); assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count()); }
@CliCommand(value = "repair corrupted clean files", help = "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); FSUtils.deleteInstantFile(client.getFs(), 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); FSUtils.deleteInstantFile(client.getFs(), client.getMetaPath(), instant); } else { throw new HoodieIOException(ioe.getMessage(), ioe); } } }); }
RepairsCommand implements CommandMarker { @CliCommand(value = "repair corrupted clean files", help = "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); FSUtils.deleteInstantFile(client.getFs(), 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); FSUtils.deleteInstantFile(client.getFs(), client.getMetaPath(), instant); } else { throw new HoodieIOException(ioe.getMessage(), ioe); } } }); } }
RepairsCommand implements CommandMarker { @CliCommand(value = "repair corrupted clean files", help = "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); FSUtils.deleteInstantFile(client.getFs(), 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); FSUtils.deleteInstantFile(client.getFs(), client.getMetaPath(), instant); } else { throw new HoodieIOException(ioe.getMessage(), ioe); } } }); }  }
RepairsCommand implements CommandMarker { @CliCommand(value = "repair corrupted clean files", help = "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); FSUtils.deleteInstantFile(client.getFs(), 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); FSUtils.deleteInstantFile(client.getFs(), client.getMetaPath(), instant); } else { throw new HoodieIOException(ioe.getMessage(), ioe); } } }); }  @CliCommand(value = "repair deduplicate", help = "De-duplicate a partition path contains duplicates & produce repaired files to replace with") String deduplicate(
      @CliOption(key = {"duplicatedPartitionPath"}, help = "Partition Path containing the duplicates",
          mandatory = true) final String duplicatedPartitionPath,
      @CliOption(key = {"repairedOutputPath"}, help = "Location to place the repaired files",
          mandatory = true) final String repairedOutputPath,
      @CliOption(key = {"sparkProperties"}, help = "Spark Properties File Path",
          unspecifiedDefaultValue = "") String sparkPropertiesPath,
      @CliOption(key = "sparkMaster", unspecifiedDefaultValue = "", help = "Spark Master") String master,
      @CliOption(key = "sparkMemory", unspecifiedDefaultValue = "4G",
          help = "Spark executor memory") final String sparkMemory,
      @CliOption(key = {"dryrun"},
          help = "Should we actually remove duplicates or just run and store result to repairedOutputPath",
          unspecifiedDefaultValue = "true") final boolean dryRun,
      @CliOption(key = {"dedupeType"}, help = "Valid values are - insert_type, update_type and upsert_type",
          unspecifiedDefaultValue = "insert_type") final String dedupeType); @CliCommand(value = "repair addpartitionmeta", help = "Add partition metadata to a table, if not present") String addPartitionMeta(
      @CliOption(key = {"dryrun"}, help = "Should we actually add or just print what would be done",
          unspecifiedDefaultValue = "true") final boolean dryRun); @CliCommand(value = "repair overwrite-hoodie-props", help = "Overwrite hoodie.properties with provided file. Risky operation. Proceed with caution!") String overwriteHoodieProperties(
      @CliOption(key = {"new-props-file"}, help = "Path to a properties file on local filesystem to overwrite the table's hoodie.properties with")
      final String overwriteFilePath); @CliCommand(value = "repair corrupted clean files", help = "repair corrupted clean files") void removeCorruptedPendingCleanAction(); }
RepairsCommand implements CommandMarker { @CliCommand(value = "repair corrupted clean files", help = "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); FSUtils.deleteInstantFile(client.getFs(), 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); FSUtils.deleteInstantFile(client.getFs(), client.getMetaPath(), instant); } else { throw new HoodieIOException(ioe.getMessage(), ioe); } } }); }  @CliCommand(value = "repair deduplicate", help = "De-duplicate a partition path contains duplicates & produce repaired files to replace with") String deduplicate(
      @CliOption(key = {"duplicatedPartitionPath"}, help = "Partition Path containing the duplicates",
          mandatory = true) final String duplicatedPartitionPath,
      @CliOption(key = {"repairedOutputPath"}, help = "Location to place the repaired files",
          mandatory = true) final String repairedOutputPath,
      @CliOption(key = {"sparkProperties"}, help = "Spark Properties File Path",
          unspecifiedDefaultValue = "") String sparkPropertiesPath,
      @CliOption(key = "sparkMaster", unspecifiedDefaultValue = "", help = "Spark Master") String master,
      @CliOption(key = "sparkMemory", unspecifiedDefaultValue = "4G",
          help = "Spark executor memory") final String sparkMemory,
      @CliOption(key = {"dryrun"},
          help = "Should we actually remove duplicates or just run and store result to repairedOutputPath",
          unspecifiedDefaultValue = "true") final boolean dryRun,
      @CliOption(key = {"dedupeType"}, help = "Valid values are - insert_type, update_type and upsert_type",
          unspecifiedDefaultValue = "insert_type") final String dedupeType); @CliCommand(value = "repair addpartitionmeta", help = "Add partition metadata to a table, if not present") String addPartitionMeta(
      @CliOption(key = {"dryrun"}, help = "Should we actually add or just print what would be done",
          unspecifiedDefaultValue = "true") final boolean dryRun); @CliCommand(value = "repair overwrite-hoodie-props", help = "Overwrite hoodie.properties with provided file. Risky operation. Proceed with caution!") String overwriteHoodieProperties(
      @CliOption(key = {"new-props-file"}, help = "Path to a properties file on local filesystem to overwrite the table's hoodie.properties with")
      final String overwriteFilePath); @CliCommand(value = "repair corrupted clean files", help = "repair corrupted clean files") void removeCorruptedPendingCleanAction(); static final String DEDUPLICATE_RETURN_PREFIX; }