#!/bin/bash

FQDN=$(hostname --fqdn)
KAFKA_LIB="/usr/lib/kafka"
# Assume we have kafka up and running on the FQDN, port 9092
KAFKA_SERVER="${FQDN}:9092"
NUM_MESSAGES=10
TEST_TOPIC="topics-test"

assert_installed() {
    which $1 > /dev/null
    if [ $? -ne 0 ]; then
	echo "$1 not installed" >&2
	exit 1
    fi
}

assert_help() {
    # Can't test for exit code, as --help can be implemented to return
    # non-zero exit codes sometimes (i.e. kafka-topics --help)
    line_count=$($1 --help 2>&1 | wc -l)
    if [ ${line_count} -lt 5 ]; then
	echo "$1 failed to display help" >&2
	exit 1
    fi
}

result=0

echo "Checking scripts are installed"
assert_installed kafka-topics
assert_installed kafka-console-producer
assert_installed kafka-console-consumer

echo "Checking scripts display help"
assert_help kafka-topics
assert_help kafka-console-producer
assert_help kafka-console-consumer

# Try to create a topic without security setup properly
TEST_TOPIC="producer-consumer-test"
echo "Creating topic without proper SSL certs"
kafka-topics --create --bootstrap-server ${KAFKA_SERVER} \
	     --replication-factor 1 \
	     --partitions 1 \
	     --topic ${TEST_TOPIC} \
	     --command-config ${KAFKA_LIB}/config/client.properties
if [ $? == 0 ]; then
    echo "Success creating topic, expected failure" >&2
    exit 1
fi

# Set up security
echo "Running security setup scripts"
printf 'y\ny\nn\n' | sudo bash ${KAFKA_LIB}/bin/kafka-ssl-keys.sh
if [ $? -ne 0 ]; then
    echo "kafka-ssl-keys.sh could not run successfully" >&2
    exit 1
fi

sudo bash ${KAFKA_LIB}/bin/kafka-zookeeper-sasl-setup.sh
if [ $? -ne 0 ]; then
    echo "kafka-zookeeper-sasl-setup.sh could not run successfully" >&2
    exit 1
fi

# Create a topic
TEST_TOPIC="producer-consumer-test"
echo "Creating topic"
kafka-topics --create --bootstrap-server ${KAFKA_SERVER} \
	     --replication-factor 1 \
	     --partitions 1 \
	     --topic ${TEST_TOPIC} \
	     --command-config ${KAFKA_LIB}/config/client.properties
if [ $? != 0 ]; then
    echo "Failure creating topic" >&2
    exit 1
fi

echo "Check listing topic"
kafka-topics --list \
             --bootstrap-server ${KAFKA_SERVER} \
	     --command-config ${KAFKA_LIB}/config/client.properties \
	     | grep ^"${TEST_TOPIC}"
if [ $? -ne 0 ]; then
    echo "Failure listing topics" >&2
    result=1
    # Fallthrough to deleting topic, for cleanup
fi

# Send some messages
for i in $(seq 1 ${NUM_MESSAGES}); do
    echo "test message ${i}" \
	| kafka-console-producer --broker-list ${KAFKA_SERVER} \
				 --topic ${TEST_TOPIC} \
			       	 --producer.config ${KAFKA_LIB}/config/client.properties
    if [ $? != 0 ]; then
	# Fallthrough to cleanup
	echo "Failure sending message" >&2
	result=1
    fi
done
echo

# Consumer
echo "Running consumer to read messages"
kafka-console-consumer --bootstrap-server ${KAFKA_SERVER} \
		       --topic ${TEST_TOPIC} \
		       --max-messages ${NUM_MESSAGES} \
		       --consumer.config ${KAFKA_LIB}/config/client.properties \
		       --from-beginning \
		       2>&1
if [ $? != 0 ]; then
    # Fallthrough to cleanup
    echo "Failure reading back ${NUM_MESSAGES} messages" >&2
    result=1
fi

echo "Cleanup"
kafka-topics --delete \
	     --bootstrap-server ${KAFKA_SERVER} \
	     --topic ${TEST_TOPIC} \
             --command-config ${KAFKA_LIB}/config/client.properties
if [ $? != 0 ]; then
    echo "Failure deleting topic" >&2
    exit 1
fi

exit $result
