From 8cdd24707097188ca368929c5ed73b7b4d2bb140 Mon Sep 17 00:00:00 2001
From: Suyog Rao <suyog.rao@gmail.com>
Date: Mon, 17 Oct 2016 19:44:16 -0700
Subject: [PATCH] add more debuggability for kafka tests

Fixes #6072
---
 qa/integration/services/helpers.sh     |  2 ++
 qa/integration/services/kafka_setup.sh | 19 +++++++++++++++----
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/qa/integration/services/helpers.sh b/qa/integration/services/helpers.sh
index 872389497..dbf970a3e 100644
--- a/qa/integration/services/helpers.sh
+++ b/qa/integration/services/helpers.sh
@@ -19,6 +19,8 @@ wait_for_port() {
         [[ $count -eq 0 ]] && return 1
         sleep 0.5
     done
+    # just in case, one more time
+    nc -z localhost $port
 }
 
 clean_install_dir() {
diff --git a/qa/integration/services/kafka_setup.sh b/qa/integration/services/kafka_setup.sh
index 0a836814b..3a3b28330 100755
--- a/qa/integration/services/kafka_setup.sh
+++ b/qa/integration/services/kafka_setup.sh
@@ -14,6 +14,7 @@ fi
 KAFKA_HOME=$INSTALL_DIR/kafka
 KAFKA_TOPIC=logstash_topic_plain
 KAFKA_MESSAGES=37
+KAFKA_LOGS_DIR=/tmp/ls_integration/kafka-logs
 
 setup_kafka() {
     local version=$1
@@ -30,28 +31,38 @@ start_kafka() {
     $KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties
     wait_for_port 2181
     echo "Starting Kafka broker"
-    $KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties --override delete.topic.enable=true --override log.dirs=/tmp/ls_integration/kafka-logs
+    $KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties --override delete.topic.enable=true --override log.dirs=$KAFKA_LOGS_DIR --override log.flush.interval.ms=200
     wait_for_port 9092
 }
 
 wait_for_messages() {
     local count=10
     local read_lines=0
+    
+    echo "Checking if Kafka topic has been populated with data"
     while [[ $read_lines -ne $KAFKA_MESSAGES ]] && [[ $count -ne 0 ]]; do
-        read_lines=`$KAFKA_HOME/bin/kafka-console-consumer.sh --topic $KAFKA_TOPIC --new-consumer --bootstrap-server localhost:9092 --from-beginning --max-messages $KAFKA_MESSAGES | wc -l`
+        read_lines=`$KAFKA_HOME/bin/kafka-console-consumer.sh --topic $KAFKA_TOPIC --new-consumer --bootstrap-server localhost:9092 --from-beginning --max-messages $KAFKA_MESSAGES --timeout-ms 10000 | wc -l`
         count=$(( $count - 1 ))
         [[ $count -eq 0 ]] && return 1
         sleep 0.5
+        ls -lrt $KAFKA_LOGS_DIR/$KAFKA_TOPIC-0/
     done
+    echo "Kafka topic has been populated with test data"
 }
 
 setup_install_dir
 setup_kafka $version
 start_kafka
 # Set up topics
-$KAFKA_HOME/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic logstash_topic_plain --zookeeper localhost:2181
+$KAFKA_HOME/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic $KAFKA_TOPIC --zookeeper localhost:2181
+# check topic got created
+num_topic=`$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper localhost:2181 | grep $KAFKA_TOPIC | wc -l`
+[[ $num_topic -eq 1 ]]
 # Add test messages to the newly created topic
-cat $current_dir/../fixtures/how_sample.input | $KAFKA_HOME/bin/kafka-console-producer.sh --topic $KAFKA_TOPIC --broker-list localhost:9092
+cp $current_dir/../fixtures/how_sample.input $KAFKA_HOME
+[[ ! -s  how_sample.input ]]
+$KAFKA_HOME/bin/kafka-console-producer.sh --topic $KAFKA_TOPIC --broker-list localhost:9092 < $KAFKA_HOME/how_sample.input
+echo "Kafka load status code $?"
 # Wait until broker has all messages
 wait_for_messages
 echo "Kafka Setup complete"