private void configureCommonProperties(TableDesc tableDesc, Map<String, String> jobProperties) {
    String topic = tableDesc.getProperties().getProperty(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName(), "");
    if (topic.isEmpty()) {
      throw new IllegalArgumentException("Kafka topic missing set table property->"
          + KafkaTableProperties.HIVE_KAFKA_TOPIC.getName());
    }
    jobProperties.put(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName(), topic);
    String
        brokerString =
        tableDesc.getProperties().getProperty(KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName(), "");
    if (brokerString.isEmpty()) {
      throw new IllegalArgumentException("Broker address missing set table property->"
          + KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName());
    }
    jobProperties.put(KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName(), brokerString);
    Arrays.stream(KafkaTableProperties.values())
        .filter(tableProperty -> !tableProperty.isMandatory())
        .forEach(tableProperty -> jobProperties.put(tableProperty.getName(),
            tableDesc.getProperties().getProperty(tableProperty.getName())));
    // If the user ask for EOS then set the read to only committed.
    if (jobProperties.get(KafkaTableProperties.WRITE_SEMANTIC_PROPERTY.getName())
        .equals(KafkaOutputFormat.WriteSemantic.EXACTLY_ONCE.name())) {
      jobProperties.put("kafka.consumer.isolation.level", "read_committed");
    }
  }
Kafka prefix to form the URI for authentication
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java