@Override public StorageHandlerInfo getStorageHandlerInfo(Table table) throws MetaException {
    String topic = table.getParameters().get(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName());
    if (topic == null || topic.isEmpty()) {
      throw new MetaException("topic is null or empty");
    }
    String brokers = table.getParameters().get(KafkaTableProperties.HIVE_KAFKA_BOOTSTRAP_SERVERS.getName());
    if (brokers == null || brokers.isEmpty()) {
      throw new MetaException("kafka brokers string is null or empty");
    }
    final Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
    properties.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokers);
    properties.setProperty(CommonClientConfigs.CLIENT_ID_CONFIG, Utilities.getTaskId(getConf()));
    if (UserGroupInformation.isSecurityEnabled()) {
      KafkaUtils.addKerberosJaasConf(getConf(), properties);
    }
    table.getParameters()
        .entrySet()
        .stream()
        .filter(objectObjectEntry -> objectObjectEntry.getKey()
            .toLowerCase()
            .startsWith(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX))
        .forEach(entry -> {
          String key = entry.getKey().substring(KafkaUtils.CONSUMER_CONFIGURATION_PREFIX.length() + 1);
          if (KafkaUtils.FORBIDDEN_PROPERTIES.contains(key)) {
            throw new IllegalArgumentException("Not suppose to set Kafka Property " + key);
          }
          properties.put(key, entry.getValue());
        });
    return new KafkaStorageHandlerInfo(topic, properties);
  }
Kafka prefix to form the URI for authentication
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java