ForwardingBigQueryFileOutputCommitter extends OutputCommitter { protected void cleanup(JobContext context) throws IOException { boolean delete = BigQueryOutputConfiguration.getCleanupTemporaryDataFlag(context.getConfiguration()); if (delete && outputFileSystem.exists(outputPath)) { logger.atInfo().log("Found GCS output data at '%s', attempting to clean up.", outputPath); if (outputFileSystem.delete(outputPath, true)) { logger.atInfo().log("Successfully deleted GCS output path '%s'.", outputPath); } else { logger.atWarning().log( "Failed to delete GCS output at '%s', retrying on shutdown.", outputPath); outputFileSystem.deleteOnExit(outputPath); } } } ForwardingBigQueryFileOutputCommitter(TaskAttemptContext context, OutputCommitter delegate); @Override void commitJob(JobContext context); @Override void abortJob(JobContext context, State state); @Override void abortTask(TaskAttemptContext context); @Override void commitTask(TaskAttemptContext context); @Override boolean needsTaskCommit(TaskAttemptContext context); @Override void setupJob(JobContext context); @Override void setupTask(TaskAttemptContext context);  }
@Test public void testCleanup() throws IOException { generateSampleFiles(); committer.cleanup(job); assertThat(!ghfs.exists(outputPath)).isTrue(); assertThat(!ghfs.exists(outputSampleFilePath)).isTrue(); }