// ConsumeTopic start consumer group with given worker and pool size
ConsumeTopic start consumer group with given worker and pool size
// ConsumeTopic start consumer group with given worker and pool size
func (c *consumerGroup) ConsumeTopic(ctx context.Context, groupTopics []string, poolSize int, worker Worker) {
	r := c.GetNewKafkaReader(c.Brokers, groupTopics, c.GroupID)
	defer func() {
		if err := r.Close(); err != nil {
			c.log.Warnf("consumerGroup.r.Close: %v", err)
		}
	}()
	c.log.Infof("Starting consumer groupID: %s, topic: %+v, pool size: %v", c.GroupID, groupTopics, poolSize)
	wg := &sync.WaitGroup{}
	for i := 0; i <= poolSize; i++ {
		wg.Add(1)
		go worker(ctx, r, wg, i)
	}
	wg.Wait()
}
func ( c * consumerGroup ) ConsumeTopic ( ctx context . Context , groupTopics [ ] string , poolSize int , worker Worker ) { r := c . GetNewKafkaReader ( c . Brokers , groupTopics , c . GroupID ) defer func ( ) { if err := r . Close ( ) ; err != nil { c . log . Warnf ( " " , err ) } } ( ) c . log . Infof ( " " , c . GroupID , groupTopics , poolSize ) wg := & sync . WaitGroup { } for i := 0 ; i <= poolSize ; i ++ { wg . Add ( 1 ) go worker ( ctx , r , wg , i ) } wg . Wait ( ) }
ConsumeTopic start consumer group with given worker and pool size
ConsumeTopic start consumer group with given worker and pool size