class IndirectBigQueryOutputCommitter extends ForwardingBigQueryFileOutputCommitter {

  @Override
  public void commitJob(JobContext context) throws IOException {
    super.commitJob(context);

    // Get the destination configuration information.
    Configuration conf = context.getConfiguration();
    TableReference destTable = BigQueryOutputConfiguration.getTableReference(conf);
    String jobProjectId = BigQueryOutputConfiguration.getJobProjectId(conf);
    String writeDisposition = BigQueryOutputConfiguration.getWriteDisposition(conf);
    String createDisposition = BigQueryOutputConfiguration.getCreateDisposition(conf);
    Optional<BigQueryTableSchema> destSchema = BigQueryOutputConfiguration.getTableSchema(conf);
    Optional<BigQueryTimePartitioning> timePartitioning =
        BigQueryOutputConfiguration.getTablePartitioning(conf);
    String kmsKeyName = BigQueryOutputConfiguration.getKmsKeyName(conf);
    BigQueryFileFormat outputFileFormat = BigQueryOutputConfiguration.getFileFormat(conf);
    List<String> sourceUris = getOutputFileURIs();

    try {
      getBigQueryHelper()
          .importFromGcs(
              jobProjectId,
              destTable,
              destSchema.isPresent() ? destSchema.get().get() : null,
              timePartitioning.isPresent() ? timePartitioning.get().get() : null,
              kmsKeyName,
              outputFileFormat,
              createDisposition,
              writeDisposition,
              sourceUris,
              true);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new IOException("Failed to import GCS into BigQuery", e);
    }

    cleanup(context);
  }

  public  IndirectBigQueryOutputCommitter(TaskAttemptContext context, OutputCommitter delegate);

}