#!/usr/bin/env bash
#
# Copyright 2016-2024 Confluent Inc.
#

set -o nounset \
    -o errexit \
    -o verbose

echo "===> Check that log dirs are writable ..."
# You need to set KAFKA_LOG_DIRS to be a child of the mount point
# We create the data directory here if necessary to avoid the issue
# where a mounted volume has a lost+found directory
# which makes the broker choke thinking it's a Kafka topic
export KAFKA_LOG_DIRS=$(grep "log.dirs" ${KAFKA_CONFIG_DIR}/kafka.properties | cut -d '=' -f 2)
printf "KAFKA_LOG_DIRS=%s\n" "$KAFKA_LOG_DIRS"
if [ ! -e "$KAFKA_LOG_DIRS" ]; then
    mkdir -p "$KAFKA_LOG_DIRS"
fi
if [ ! -w "$KAFKA_LOG_DIRS" ]; then
    echo "ERROR: $KAFKA_LOG_DIRS is not writable" >&2
    exit 1
fi

echo "===> Check that log4j dir is writable ..."
if [ ! -w "$KAFKA_LOG4J_DIR" ]; then
    echo "ERROR: $KAFKA_LOG4J_DIR is not writable" >&2
    exit 1
fi

kraft_cluster_id=$(grep "^#KRaftClusterId=" ${KAFKA_CONFIG_DIR}/kafka.properties | cut -d '=' -f 2)
export KAFKA_ZOOKEEPER_CONNECT=$(grep "zookeeper.connect" ${KAFKA_CONFIG_DIR}/kafka.properties | cut -d '=' -f 2)
if [ -z "$kraft_cluster_id" -a -z "$KAFKA_ZOOKEEPER_CONNECT" ]; then
    echo "Fail: No zookeeper.connect and no #KRaftClusterId in ${KAFKA_CONFIG_DIR}/kafka.properties"
    exit 1
fi
if [ ! -z "$kraft_cluster_id" ]; then
    # KMETA-405: invoke with --release-version if a string value is given,
    # otherwise convert the integer value given to a string value first
    kraft_release_version=$(grep "^#KRaftReleaseVersion=" ${KAFKA_CONFIG_DIR}/kafka.properties | cut -d '=' -f 2)
    if [ -z "$kraft_release_version" ]; then
        confluent_metadata_version=$(grep "^#ConfluentMetadataVersion=" ${KAFKA_CONFIG_DIR}/kafka.properties | cut -d '=' -f 2)
        case $confluent_metadata_version in
          1) kraft_release_version=CP-7.0-IV1;;
          2) kraft_release_version=CP-7.1-IV0;;
          3) kraft_release_version=CP-7.2-IV0;;
          4) kraft_release_version=CP-7.3-IV0;;
          5) kraft_release_version=CP-7.3-IV1;;
          6) kraft_release_version=CP-7.3-IV2;;
          7) kraft_release_version=CP-7.3-IV3;;
          107) kraft_release_version=CP-7.3-IV3;;
          108) kraft_release_version=CP-7.4-IV0;;
          109) kraft_release_version=CP-7.5-IV0;;
          110) kraft_release_version=CP-7.5-IV1;;
          111) kraft_release_version=CP-7.5-IV2;;
          112) kraft_release_version=CP-7.6-IV0;;
          113) kraft_release_version=CP-7.6-IV1;;
          114) kraft_release_version=CP-7.6-IV2;;
          115) kraft_release_version=CP-7.7-IV0;;
          116) kraft_release_version=CP-7.7-IV1;;
          117) kraft_release_version=CP-7.7-IV2;;
          118) kraft_release_version=CP-7.7-IV3;;
          119) kraft_release_version=CP-7.7-IV4;;
          120) kraft_release_version=CP-7.8-IV0A;;
          *)
            echo "Fail: Unsupported #ConfluentMetadataVersion in ${KAFKA_CONFIG_DIR}/kafka.properties: $confluent_metadata_version"
            exit 1
            ;;
        esac
        echo "===> Converted Confluent Metadata Version $confluent_metadata_version to Release Version $kraft_release_version"
    fi
    echo "===> Run kafka-storage tool with --release-version=$kraft_release_version ..."
    /opt/confluent/bin/kafka-storage.sh format --ignore-formatted --config ${KAFKA_CONFIG_DIR}/kafka.properties --cluster-id=$kraft_cluster_id --release-version=$kraft_release_version || exit $?
fi

if [ ! -z "$KAFKA_ZOOKEEPER_CONNECT" ]; then
    echo "===> Check if Zookeeper is healthy ..."
    # Use kafka-run-class.sh to build the classpath
    /opt/confluent/bin/kafka-run-class.sh \
        io.confluent.admin.utils.cli.ZookeeperReadyCommand \
        "$KAFKA_ZOOKEEPER_CONNECT" \
        "$(expr "${KAFKA_CUB_ZK_TIMEOUT:-40}" \* 1000)" || exit $?
fi
