Improve consumer ready status

Signed-off-by: axfor <aixiaoxiang2009@hotmail.com>

cmd/ingester/app/consumer/consumer.go

@@ -53,14 +53,27 @@ type Consumer struct {
 
 	doneWg sync.WaitGroup
 
-	ready          chan struct{}
-	readyClose     bool
-	readyCloseLock sync.Mutex
+	consumerReady *ConsumerReady
 
 	topic  string
 	cancel context.CancelFunc
 }
 
+type ConsumerReady struct {
+	ready     chan struct{}
+	closeOnce sync.Once
+}
+
+func (c *ConsumerReady) Ready() {
+	<-c.ready
+}
+
+func (c *ConsumerReady) Close() {
+	c.closeOnce.Do(func() {
+		close(c.ready)
+	})
+}
+
 // New is a constructor for a Consumer
 func New(params Params) (*Consumer, error) {
 	deadlockDetector := newDeadlockDetector(params.MetricsFactory, params.Logger, params.DeadlockCheckInterval)
@@ -71,8 +84,10 @@ func New(params Params) (*Consumer, error) {
 		processorFactory:    params.ProcessorFactory,
 		deadlockDetector:    deadlockDetector,
 		partitionsHeldGauge: partitionsHeldGauge(params.MetricsFactory),
-		ready:               make(chan struct{}),
-		topic:               params.ProcessorFactory.topic,
+		consumerReady: &ConsumerReady{
+			ready: make(chan struct{}),
+		},
+		topic: params.ProcessorFactory.topic,
 	}, nil
 }
 
@@ -99,7 +114,9 @@ func (c *Consumer) Start() {
 				if ctx.Err() != nil {
 					return
 				}
-				c.ready = make(chan struct{})
+				c.consumerReady = &ConsumerReady{
+					ready: make(chan struct{}),
+				}
 			}
 		}
 	}()
@@ -125,21 +142,13 @@ func (c *Consumer) Close() error {
 
 // Ready is consumer running
 func (c *Consumer) Ready() {
-	<-c.ready
+	c.consumerReady.Ready()
 }
 
 // Setup is run at the beginning of a new session, before ConsumeClaim
 func (c *Consumer) Setup(sarama.ConsumerGroupSession) error {
 	// Mark the consumer as ready
-
-	c.readyCloseLock.Lock()
-	defer c.readyCloseLock.Unlock()
-
-	if !c.readyClose {
-		c.readyClose = true
-		close(c.ready)
-	}
-
+	c.consumerReady.Close()
 	return nil
 }