@Override
  public void storageHandlerCommit(Properties commitProperties, Operation operation, ExecutorService executorService)
      throws HiveException {
    String tableName = commitProperties.getProperty(Catalogs.NAME);
    String location = commitProperties.getProperty(Catalogs.LOCATION);
    String snapshotRef = commitProperties.getProperty(Catalogs.SNAPSHOT_REF);
    Configuration configuration = SessionState.getSessionConf();
    if (location != null) {
      HiveTableUtil.cleanupTableObjectFile(location, configuration);
    }
    List<JobContext> jobContextList = HiveIcebergOutputCommitter
            .generateJobContext(configuration, tableName, snapshotRef);
    if (jobContextList.isEmpty()) {
      return;
    }
    HiveIcebergOutputCommitter committer = getOutputCommitter();
    committer.setWorkerPool(executorService);
    try {
      committer.commitJobs(jobContextList, operation);
    } catch (Throwable e) {
      String ids = jobContextList
          .stream().map(jobContext -> jobContext.getJobID().toString()).collect(Collectors.joining(", "));
      // Aborting the job if the commit has failed
      LOG.error("Error while trying to commit job: {}, starting rollback changes for table: {}",
          ids, tableName, e);
      try {
        committer.abortJobs(jobContextList);
      } catch (IOException ioe) {
        LOG.error("Error while trying to abort failed job. There might be uncleaned data files.", ioe);
        // no throwing here because the original exception should be propagated
      }
      throw new HiveException(
          "Error committing job: " + ids + " for table: " + tableName, e);
    }
  }
Iceberg's optimistic concurrency control fails to provide means for IOW and Insert operations isolation.
Use `hive.txn.ext.locking.enabled` config to create Hive locks in order to guarantee data consistency.
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java