From 7c1a3b12a56f2e659e55af89d7b4ed2e5cd84f2e Mon Sep 17 00:00:00 2001 From: Doug Davis Date: Mon, 23 Oct 2023 13:22:32 +0000 Subject: [PATCH] fix race Signed-off-by: Doug Davis --- test/integration/kafka_sarama_binding/kafka_test.go | 2 +- v2/client/test/test.go | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/test/integration/kafka_sarama_binding/kafka_test.go b/test/integration/kafka_sarama_binding/kafka_test.go index 372701c93..6c5851113 100644 --- a/test/integration/kafka_sarama_binding/kafka_test.go +++ b/test/integration/kafka_sarama_binding/kafka_test.go @@ -104,7 +104,7 @@ func testSenderReceiver(t testing.TB) (func(), bindings.Sender, bindings.Receive // Not perfect but we need to give OpenInbound() as chance to start // as it's a race condition. I couldn't find something on 'p' to wait for - time.Sleep(5 * time.Second) + time.Sleep(15 * time.Second) return func() { require.NoError(t, p.Close(context.TODO())) diff --git a/v2/client/test/test.go b/v2/client/test/test.go index 5124cfd6c..8fd5be941 100644 --- a/v2/client/test/test.go +++ b/v2/client/test/test.go @@ -10,6 +10,7 @@ import ( "context" "sync" "testing" + "time" "github.com/cloudevents/sdk-go/v2/protocol" @@ -29,6 +30,9 @@ func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Even wg := sync.WaitGroup{} wg.Add(2) + // Give time for Kafka client protocol to get setup + time.Sleep(5 * time.Second) + go func() { ctx, cancel := context.WithCancel(context.TODO()) inCh := make(chan event.Event) @@ -49,6 +53,9 @@ func SendReceive(t *testing.T, protocolFactory func() interface{}, in event.Even outAssert(e) }() + // Give time for the receiever to start + time.Sleep(5 * time.Second) + go func() { defer wg.Done() err := c.Send(context.Background(), in)