@Override
  public List<InputSplit> getSplits(JobContext context) {
    Configuration conf = context.getConfiguration();
    Table table = Optional
        .ofNullable(HiveIcebergStorageHandler.table(conf, conf.get(InputFormatConfig.TABLE_IDENTIFIER)))
        .orElseGet(() -> {
          Table tbl = Catalogs.loadTable(conf);
          conf.set(InputFormatConfig.TABLE_IDENTIFIER, tbl.name());
          conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tbl.name(), SerializationUtil.serializeToBase64(tbl));
          return tbl;
        });
    final ExecutorService workerPool =
        ThreadPools.newWorkerPool("iceberg-plan-worker-pool",
            conf.getInt(SystemConfigs.WORKER_THREAD_POOL_SIZE.propertyKey(), ThreadPools.WORKER_THREAD_POOL_SIZE));
    try {
      return planInputSplits(table, conf, workerPool);
    } finally {
      workerPool.shutdown();
    }
  }
Configures the {@code Job} to use the {@code IcebergInputFormat} and
returns a helper to add further configuration.

@param job the {@code Job} to configure
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java