fixed some err of integration test

Signed-off-by: axfor <aixiaoxiang2009@hotmail.com>
axfor · Mar 18, 2023 · 7294b58 · 7294b58
1 parent 90ddffc
commit 7294b58
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 0 deletions.
diff --git a/cmd/ingester/app/consumer/consumer.go b/cmd/ingester/app/consumer/consumer.go
@@ -77,7 +77,11 @@ func (c *Consumer) Start() {
 	ctx, cancel := context.WithCancel(context.Background())
 	c.cancel = cancel
 	c.doneWg.Add(1)
+
+	var firstStart sync.WaitGroup
+	firstStart.Add(1)
 	go func() {
+		firstStart.Done()
 		defer c.doneWg.Done()
 		for {
 			select {
@@ -92,6 +96,7 @@ func (c *Consumer) Start() {
 			}
 		}
 	}()
+	firstStart.Wait()
 }
 
 // Close closes the Consumer and underlying sarama consumer

diff --git a/plugin/storage/integration/kafka_test.go b/plugin/storage/integration/kafka_test.go
@@ -99,6 +99,9 @@ func (s *KafkaIntegrationTestSuite) initialize() error {
 	}
 	spanConsumer.Start()
 
+	//wait consumer running
+	time.Sleep(time.Second * 5)
+
 	s.SpanWriter = spanWriter
 	s.SpanReader = &ingester{traceStore}
 	s.Refresh = func() error { return nil }