@Override public void preCreateTable(Table table) throws MetaException {
    if (!table.getTableType().equals(TableType.EXTERNAL_TABLE.toString())) {
      throw new MetaException(KAFKA_STORAGE_HANDLER + " supports only " + TableType.EXTERNAL_TABLE);
    }
    Arrays.stream(KafkaTableProperties.values())
        .filter(KafkaTableProperties::isMandatory)
        .forEach(key -> Preconditions.checkNotNull(table.getParameters().get(key.getName()),
            "Set Table property " + key.getName()));
    // Put all the default at the pre create.
    Arrays.stream(KafkaTableProperties.values()).forEach((key) -> {
      if (table.getParameters().get(key.getName()) == null) {
        table.putToParameters(key.getName(), key.getDefaultValue());
      }
    });
  }
Kafka prefix to form the URI for authentication
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaStorageHandler.java