@Override
  public void abortJob(JobContext originalContext, int status) throws IOException {
    JobContext jobContext = TezUtil.enrichContextWithVertexId(originalContext);
    JobConf jobConf = jobContext.getJobConf();

    LOG.info("Job {} is aborted. Data file cleaning started", jobContext.getJobID());
    Collection<String> outputs = HiveIcebergStorageHandler.outputTables(jobContext.getJobConf());
    Collection<String> jobLocations = new ConcurrentLinkedQueue<>();

    ExecutorService fileExecutor = fileExecutor(jobConf);
    ExecutorService tableExecutor = tableExecutor(jobConf, outputs.size());
    try {
      // Cleans up the changes for the output tables in parallel
      Tasks.foreach(outputs)
          .suppressFailureWhenFinished()
          .executeWith(tableExecutor)
          .onFailure((output, exc) -> LOG.warn("Failed cleanup table {} on abort job", output, exc))
          .run(
              output -> {
                LOG.info("Cleaning table {} with job id {}", output, jobContext.getJobID());
                Table table = HiveIcebergStorageHandler.table(jobConf, output);
                jobLocations.add(
                    generateJobLocation(table.location(), jobConf, jobContext.getJobID()));
                Collection<DataFile> dataFiles =
                    dataFiles(fileExecutor, table.location(), jobContext, table.io(), false);

                // Check if we have files already committed and remove data files if there are any
                if (!dataFiles.isEmpty()) {
                  Tasks.foreach(dataFiles)
                      .retry(3)
                      .suppressFailureWhenFinished()
                      .executeWith(fileExecutor)
                      .onFailure(
                          (file, exc) ->
                              LOG.warn(
                                  "Failed to remove data file {} on abort job", file.path(), exc))
                      .run(file -> table.io().deleteFile(file.path().toString()));
                }
              });
    } finally {
      fileExecutor.shutdown();
      if (tableExecutor != null) {
        tableExecutor.shutdown();
      }
    }

    LOG.info("Job {} is aborted. Data file cleaning finished", jobContext.getJobID());

    cleanup(jobContext, jobLocations);
  }
@Test
  public void testAbortJob() throws IOException {
    HiveIcebergOutputCommitter committer = new HiveIcebergOutputCommitter();
    Table table = table(temp.toFile().getPath(), false);
    JobConf conf = jobConf(table, 1);
    writeRecords(table.name(), 1, 0, true, false, conf);
    committer.abortJob(new JobContextImpl(conf, JOB_ID), JobStatus.State.FAILED);

    HiveIcebergTestUtils.validateFiles(table, conf, JOB_ID, 0);
    HiveIcebergTestUtils.validateData(table, Collections.emptyList(), 0);
  }