@Override public void close(boolean abort) throws IOException {
    if (abort) {
      LOG.info("Aborting is set to TRUE, Closing writerId [{}] without flush.", writerId);
      producer.close(Duration.ofMillis(0));
      return;
    } else {
      LOG.info("Flushing Kafka Producer with writerId [{}]", writerId);
      producer.flush();
      LOG.info("Closing WriterId [{}]", writerId);
      producer.close();
    }
    LOG.info("Closed WriterId [{}] Delivery semantic [{}], Topic[{}], Total sent Records [{}], Total Lost Records [{}]",
        writerId, writeSemantic,
        topic,
        sentRecords,
        lostRecords.get());
    checkExceptions();
  }
@param topic Kafka Topic.
@param writerId Writer Id use for logging.
@param properties Kafka Producer properties.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/SimpleKafkaWriter.java