Skip to content

Commit

Permalink
Fixes some flaky tests in the build as well as the case when tests st…
Browse files Browse the repository at this point in the history
…art before kafka is ready.
  • Loading branch information
nachogiljaldo committed Nov 15, 2024
1 parent a8e5eab commit de34fb8
Show file tree
Hide file tree
Showing 5 changed files with 39 additions and 7 deletions.
3 changes: 3 additions & 0 deletions .circleci/config.yml
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ jobs:
key: kafka-go-mod-{{ checksum "go.sum" }}-1
paths:
- /go/pkg/mod
- run:
name: Wait for kafka
command: ./scripts/wait-for-kafka.sh
- run:
name: Test kafka-go
command: go test -race -cover ./...
Expand Down
2 changes: 2 additions & 0 deletions addoffsetstotxn_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
19 changes: 12 additions & 7 deletions reader_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -309,7 +309,7 @@ func createTopic(t *testing.T, topic string, partitions int) {
ReplicationFactor: 1,
},
},
Timeout: milliseconds(time.Second),
Timeout: milliseconds(5 * time.Second),
})
if err != nil {
if !errors.Is(err, TopicAlreadyExists) {
Expand Down Expand Up @@ -364,8 +364,8 @@ func waitForTopic(ctx context.Context, t *testing.T, topic string) {
}
}

t.Logf("retrying after 1s")
time.Sleep(time.Second)
t.Logf("retrying after 100ms")
time.Sleep(100 * time.Millisecond)
continue
}
}
Expand Down Expand Up @@ -1559,17 +1559,22 @@ func TestConsumerGroupWithGroupTopicsSingle(t *testing.T) {
}
}

func TestConsumerGroupWithGroupTopicsMultple(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
func TestConsumerGroupWithGroupTopicsMultiple(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()

client, shutdown := newLocalClient()
defer shutdown()

t1 := makeTopic()
createTopic(t, t1, 1)
defer deleteTopic(t, t1)
t2 := makeTopic()
createTopic(t, t2, 1)
defer deleteTopic(t, t2)
conf := ReaderConfig{
Brokers: []string{"localhost:9092"},
GroupID: makeGroupID(),
GroupTopics: []string{makeTopic(), makeTopic()},
GroupTopics: []string{t1, t2},
MaxWait: time.Second,
PartitionWatchInterval: 100 * time.Millisecond,
WatchPartitionChanges: true,
Expand Down
20 changes: 20 additions & 0 deletions scripts/wait-for-kafka.sh
Original file line number Diff line number Diff line change
@@ -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
2 changes: 2 additions & 0 deletions txnoffsetcommit_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down

0 comments on commit de34fb8

Please sign in to comment.