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