More fixes.

Author: nachogiljaldo

.circleci/config.yml

@@ -56,7 +56,7 @@ jobs:
         - /go/pkg/mod
     - run:
         name: Wait for kafka
-        command: COUNTER=0; nc -Z localhost 9092; until [ $? -eq 0 ] || [ "$COUNTER" -ge 60 ]; do let COUNTER=$COUNTER+1; sleep 1; echo "waiting $COUNTER"; nc -Z localhost 9092; done
+        command: ./scripts/wait-for-kafka.sh
     - run:
         name: Test kafka-go
         command: go test -race -cover ./...

addoffsetstotxn_test.go

@@ -22,11 +22,13 @@ func TestClientAddOffsetsToTxn(t *testing.T) {
 	defer shutdown()
 
 	err := clientCreateTopic(client, topic, 3)
+	defer deleteTopic(t, topic)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	waitForTopic(ctx, t, topic)
 	defer cancel()
 	respc, err := waitForCoordinatorIndefinitely(ctx, client, &FindCoordinatorRequest{
 		Addr:    client.Addr,

reader_test.go

@@ -365,7 +365,7 @@ func waitForTopic(ctx context.Context, t *testing.T, topic string) {
 		}
 
 		t.Logf("retrying after 1s")
-		time.Sleep(time.Second)
+		time.Sleep(100 * time.Second)
 		continue
 	}
 }

scripts/wait-for-kafka.sh

@@ -0,0 +1,20 @@
+#/bin/bash
+
+COUNTER=0; 
+echo foo | nc localhost 9092
+STATUS=$?
+ATTEMPTS=60
+until [ ${STATUS} -eq 0 ] || [ "$COUNTER" -ge "${ATTEMPTS}" ];
+do
+    let COUNTER=$COUNTER+1;
+    sleep 1;
+    echo "[$COUNTER] waiting for 9092 port to be open";
+    echo foo | nc localhost 9092
+    STATUS=$?
+done
+
+if [ "${COUNTER}" -gt "${ATTEMPTS}" ];
+then
+    echo "Kafka is not running, failing"
+    exit 1
+fi

txnoffsetcommit_test.go

@@ -21,6 +21,8 @@ func TestClientTxnOffsetCommit(t *testing.T) {
 
 	client, shutdown := newLocalClientWithTopic(topic, 1)
 	defer shutdown()
+	waitForTopic(context.TODO(), t, topic)
+	defer deleteTopic(t, topic)
 
 	now := time.Now()