@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;
      }
Kafka prefix to form the URI for authentication
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java