1 files changed, 60 insertions, 0 deletions
diff --git a/tests/kafka-selftest.sh b/tests/kafka-selftest.sh
new file mode 100755
index 0000000..9b38fe0
--- /dev/null
+++ b/tests/kafka-selftest.sh
@@ -0,0 +1,60 @@
+#!/bin/bash
+# added 2018-10-26 by Rainer Gerhards
+# This file is part of the rsyslog project, released under ASL 2.0
+. ${srcdir:=.}/diag.sh init
+check_command_available kafkacat
+export KEEP_KAFKA_RUNNING="YES"
+
+export TESTMESSAGES=100000
+
+export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
+
+# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
+#export EXTRA_EXITCHECK=dumpkafkalogs
+export EXTRA_EXIT=kafka
+download_kafka
+stop_zookeeper
+stop_kafka
+
+start_zookeeper
+start_kafka
+create_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
+
+printf 'injecting messages via kafkacat\n'
+injectmsg_kafkacat
+
+# experimental: wait until kafkacat receives everything
+
+timeoutend=10
+timecounter=0
+
+printf 'receiving messages via kafkacat\n'
+while [ $timecounter -lt $timeoutend ]; do
+	(( timecounter++ ))
+
+	kafkacat -b localhost:29092 -e -C -o beginning -t $RANDTOPIC -f '%s\n' > $RSYSLOG_OUT_LOG
+	count=$(wc -l < ${RSYSLOG_OUT_LOG})
+	if [ $count -eq $TESTMESSAGES ]; then
+		printf '**** wait-kafka-lines success, have %d lines ****\n\n' "$TESTMESSAGES"
+		break
+	else
+		if [ "x$timecounter" == "x$timeoutend" ]; then
+			echo wait-kafka-lines failed, expected $TESTMESSAGES got $count
+			error_exit 1
+		else
+			echo wait-file-lines not yet there, currently $count lines
+			printf '\n'
+			$TESTTOOL_DIR/msleep 1000
+		fi
+	fi
+done
+unset count
+
+#end experimental
+
+delete_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
+
+sed -i 's/ msgnum://' "$RSYSLOG_OUT_LOG"
+seq_check 1 $TESTMESSAGES -d
+
+exit_test