#!/bin/bash
# Copyright Broadcom, Inc. All Rights Reserved.
# SPDX-License-Identifier: APACHE-2.0

# shellcheck disable=SC1091

set -o errexit
set -o nounset
set -o pipefail
# set -o xtrace # Uncomment this line for debugging purposes

# Load libraries
. /opt/bitnami/scripts/libfs.sh
. /opt/bitnami/scripts/libos.sh
. /opt/bitnami/scripts/libkafka.sh

# Load Kafka environment variables
. /opt/bitnami/scripts/kafka-env.sh

# Map Kafka environment variables
kafka_create_alias_environment_variables

# Dynamically set node.id/broker.id/controller.quorum.bootstrap.servers if the _COMMAND environment variable is set
kafka_dynamic_environment_variables
# Set the default truststore locations before validation
kafka_configure_default_truststore_locations
# Ensure Kafka user and group exist when running as 'root'
am_i_root && ensure_user_exists "$KAFKA_DAEMON_USER" --group "$KAFKA_DAEMON_GROUP"
# Ensure directories used by Kafka exist and have proper ownership and permissions
for dir in "$KAFKA_LOG_DIR" "$KAFKA_CONF_DIR" "$KAFKA_MOUNTED_CONF_DIR" "$KAFKA_VOLUME_DIR" "$KAFKA_DATA_DIR"; do
    if am_i_root; then
        ensure_dir_exists "$dir" "$KAFKA_DAEMON_USER" "$KAFKA_DAEMON_GROUP"
    else
        ensure_dir_exists "$dir"
    fi
done

# Kafka validation, skipped if server.properties was mounted at either $KAFKA_MOUNTED_CONF_DIR or $KAFKA_CONF_DIR
[[ ! -f "${KAFKA_MOUNTED_CONF_DIR}/server.properties" && ! -f "$KAFKA_CONF_FILE" ]] && kafka_validate
# Kafka initialization, skipped if server.properties was mounted at $KAFKA_CONF_DIR
[[ ! -f "$KAFKA_CONF_FILE" ]] && kafka_initialize

if kafka_is_zookeeper_supported; then
    # Initialize KRaft metadata storage if process.roles configured ...
    if grep -q "^process.roles=" "$KAFKA_CONF_FILE"; then
        # ... unless initialization is skipped
        ! is_boolean_yes "$KAFKA_SKIP_KRAFT_STORAGE_INIT" && kafka_kraft_storage_initialize
    fi
    # Configure Zookeeper SCRAM users
    if is_boolean_yes "${KAFKA_ZOOKEEPER_BOOTSTRAP_SCRAM_USERS:-}"; then
        kafka_zookeeper_create_sasl_scram_users
    fi
else
    # Initialize KRaft metadata unless initialization is skipped
    ! is_boolean_yes "$KAFKA_SKIP_KRAFT_STORAGE_INIT" && kafka_kraft_storage_initialize
fi
# Ensure custom initialization scripts are executed
kafka_custom_init_scripts
