FederatedBigQueryOutputCommitter extends ForwardingBigQueryFileOutputCommitter { @Override public void commitJob(JobContext context) throws IOException { super.commitJob(context); Configuration conf = context.getConfiguration(); TableReference destTable = BigQueryOutputConfiguration.getTableReference(conf); String jobProjectId = BigQueryOutputConfiguration.getJobProjectId(conf); Optional<BigQueryTableSchema> destSchema = BigQueryOutputConfiguration.getTableSchema(conf); BigQueryFileFormat outputFileFormat = BigQueryOutputConfiguration.getFileFormat(conf); List<String> sourceUris = getOutputFileURIs(); getBigQueryHelper() .importFederatedFromGcs( jobProjectId, destTable, destSchema.isPresent() ? destSchema.get().get() : null, outputFileFormat, sourceUris); } FederatedBigQueryOutputCommitter(TaskAttemptContext context, OutputCommitter delegate); @Override void commitJob(JobContext context); @Override void abortJob(JobContext context, State state);  }
@Test public void testCommitJob() throws IOException { generateSampleFiles(); committer.commitJob(job); @SuppressWarnings({"rawtypes", "unchecked", "cast"}) Class<List<String>> listClass = (Class<List<String>>) (Class) List.class; ArgumentCaptor<List<String>> gcsOutputFileCaptor = ArgumentCaptor.forClass(listClass); verify(mockBigQueryHelper) .importFederatedFromGcs( eq(TEST_PROJECT_ID), eq(outputTableRef), eq(TEST_TABLE_SCHEMA.get()), eq(TEST_FILE_FORMAT), gcsOutputFileCaptor.capture()); verify(mockCommitter).commitJob(eq(job)); assertThat(gcsOutputFileCaptor.getValue()).containsExactly(TEST_OUTPUT_FILE_STRING); }