private void handleKafkaException(KafkaException kafkaException) {
    if (kafkaException instanceof TimeoutException) {
      //This might happen if the producer cannot send data to the Kafka cluster and thus, its internal buffer fills up.
      LOG.error(TIMEOUT_CONFIG_HINT, kafkaException.getMessage());
    }
    if (KafkaUtils.exceptionIsFatal(kafkaException)) {
      LOG.error(String.format(ABORT_MSG, writerId, kafkaException.getMessage(), topic, -1L));
      sendExceptionRef.compareAndSet(null, kafkaException);
    } else {
      LOG.error(ACTION_ABORT, writerId, topic, writeSemantic, kafkaException.getMessage());
      sendExceptionRef.compareAndSet(null, kafkaException);
    }
  }
@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