wait

Signed-off-by: Pavol Loffay <[email protected]>

Author: pavolloffay

receiver/kafkareceiver/kafka_receiver_test.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
@@ -72,7 +73,7 @@ func TestReceiverStartConsume(t *testing.T) {
 }
 
 func TestReceiver_error(t *testing.T) {
-	zcore, o := observer.New(zapcore.ErrorLevel)
+	zcore, logObserver := observer.New(zapcore.ErrorLevel)
 	logger := zap.New(zcore)
 
 	expectedErr := fmt.Errorf("handler error")
@@ -86,7 +87,10 @@ func TestReceiver_error(t *testing.T) {
 	err := c.Start(context.Background(), nil)
 	require.NoError(t, err)
 	c.Shutdown(context.Background())
-	assert.True(t, o.FilterField(zap.Error(expectedErr)).Len() > 0)
+	waitUntil(func() bool {
+		return logObserver.FilterField(zap.Error(expectedErr)).Len() > 0
+	}, 100, time.Millisecond*100)
+	assert.True(t, logObserver.FilterField(zap.Error(expectedErr)).Len() > 0)
 }
 
 func TestConsumerGroupHandler(t *testing.T) {
@@ -278,3 +282,12 @@ func (t testConsumerGroup) Errors() <-chan error {
 func (t testConsumerGroup) Close() error {
 	return nil
 }
+
+func waitUntil(f func() bool, iterations int, sleepInterval time.Duration) {
+	for i := 0; i < iterations; i++ {
+		if f() {
+			return
+		}
+		time.Sleep(sleepInterval)
+	}
+}