@Override public void commitInsertTable(Table table, boolean overwrite) throws MetaException {
    boolean
        isExactlyOnce =
        table.getParameters()
            .get(KafkaTableProperties.WRITE_SEMANTIC_PROPERTY.getName())
            .equals(KafkaOutputFormat.WriteSemantic.EXACTLY_ONCE.name());
    String optimiticCommitVal = table.getParameters().get(KafkaTableProperties.HIVE_KAFKA_OPTIMISTIC_COMMIT.getName());
    boolean isTwoPhaseCommit = !Boolean.parseBoolean(optimiticCommitVal);
    if (!isExactlyOnce || !isTwoPhaseCommit) {
      //Case it is not 2 phase commit no open transaction to handel.
      return;
    }

    final Path queryWorkingDir = getQueryWorkingDir(table);
    final Map<String, Pair<Long, Short>> transactionsMap;
    final int maxTries = Integer.parseInt(table.getParameters().get(KafkaTableProperties.MAX_RETRIES.getName()));
    // We have 4 Stages ahead of us:
    // 1 Fetch Transactions state from HDFS.
    // 2 Build/inti all the Kafka producers and perform a pre commit call to check if we can go ahead with commit.
    // 3 Commit Transactions one by one.
    // 4 Clean workingDirectory.

    //First stage fetch the Transactions states
    final RetryUtils.Task<Map<String, Pair<Long, Short>>>
        fetchTransactionStates =
        new RetryUtils.Task<Map<String, Pair<Long, Short>>>() {
          @Override public Map<String, Pair<Long, Short>> perform() throws Exception {
            return TransactionalKafkaWriter.getTransactionsState(FileSystem.get(getConf()), queryWorkingDir);
          }
        };

    try {
      transactionsMap = RetryUtils.retry(fetchTransactionStates, (error) -> (error instanceof IOException), maxTries);
    } catch (Exception e) {
      // Can not go further
      LOG.error("Can not fetch Transaction states due [{}]", e.getMessage());
      throw new MetaException(e.getMessage());
    }

    //Second Stage Resume Producers and Pre commit
    final Properties baseProducerPros = buildProducerProperties(table);
    final Map<String, HiveKafkaProducer> producersMap = new HashMap<>();
    final RetryUtils.Task<Void> buildProducersTask = new RetryUtils.Task<Void>() {
      @Override public Void perform() throws Exception {
        assert producersMap.size() == 0;
        transactionsMap.forEach((key, value) -> {
          // Base Producer propeties, missing the transaction Id.
          baseProducerPros.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, key);
          HiveKafkaProducer<byte[], byte[]> producer = new HiveKafkaProducer<>(baseProducerPros);
          producer.resumeTransaction(value.getLeft(), value.getRight());
          // This is a dummy RPC call to ensure that the producer still resumable and signal the Pre-commit as per :
          // https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka#EndPhase
          producer.sendOffsetsToTransaction(ImmutableMap.of(), "__dry_run");
          producersMap.put(key, producer);
        });
        return null;
      }
    };

    RetryUtils.CleanupAfterFailure cleanUpTheMap = new RetryUtils.CleanupAfterFailure() {
      @Override public void cleanup() {
        producersMap.forEach((s, producer) -> producer.close(Duration.ofMillis(0)));
        producersMap.clear();
      }
    };
    final Predicate<Throwable>
        isRetrayable = (error) -> !KafkaUtils.exceptionIsFatal(error) && !(error instanceof ProducerFencedException);
    try {
      RetryUtils.retry(buildProducersTask, isRetrayable, cleanUpTheMap, maxTries, "Error while Builing Producers");
    } catch (Exception e) {
      // Can not go further
      LOG.error("Can not fetch build produces due [{}]", e);
      throw new MetaException(e.getMessage());
    }

    //Third Stage Commit Transactions, this part is the actual critical section.
    //The commit might be retried on error, but keep in mind in some cases, like open transaction can expire
    //after timeout duration of 15 mins it is not possible to go further.
    final Set<String> committedTx = new HashSet<>();
    final RetryUtils.Task<Void> commitTask = new RetryUtils.Task() {
      @Override public Object perform() throws Exception {
        producersMap.forEach((key, producer) -> {
          if (!committedTx.contains(key)) {
            producer.commitTransaction();
            committedTx.add(key);
            producer.close();
            LOG.info("Committed Transaction [{}]", key);
          }
        });
        return null;
      }
    };

    try {
      RetryUtils.retry(commitTask, isRetrayable, maxTries);
    } catch (Exception e) {
      // at this point we are in a funky state if one commit happend!! close and log it
      producersMap.forEach((key, producer) -> producer.close(Duration.ofMillis(0)));
      LOG.error("Commit transaction failed", e);
      if (committedTx.size() > 0) {
        LOG.error("Partial Data Got Commited Some actions need to be Done");
        committedTx.stream().forEach(key -> LOG.error("Transaction [{}] is an orphen commit", key));
      }
      throw new MetaException(e.getMessage());
    }

    //Stage four, clean the Query Directory
    final RetryUtils.Task<Void> cleanQueryDirTask = new RetryUtils.Task<Void>() {
      @Override public Void perform() throws Exception {
        cleanWorkingDirectory(queryWorkingDir);
        return null;
      }
    };
    try {
      RetryUtils.retry(cleanQueryDirTask, (error) -> error instanceof IOException, maxTries);
    } catch (Exception e) {
      //just log it
      LOG.error("Faild to clean Query Working Directory [{}] due to [{}]", queryWorkingDir, e.getMessage());
    }
  }
Kafka prefix to form the URI for authentication
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java