Merge pull request #35 from meitu/fix/rebalance_bug

Fix rebalance bug:
1. rebalance was triggered while the consumers changed, and close the triggerCh
2. partition consumer would give up to claim the partition when received the stop signal, then call `stop` function would reclose the triggerCh as well, and the claim would be pending
3. topic consumer wait partition consumer would be pending as well, because the partition consumer was pending

consumer_group.go

@@ -30,13 +30,13 @@ type ConsumerGroup struct {
 	topicConsumers map[string]*topicConsumer
 	saramaConsumer sarama.Consumer
 
-	id        string
-	state     int
-	wg        sync.WaitGroup
-	stopCh    chan struct{}
-	triggerCh chan int
-	stopOnce  *sync.Once
-	owners    map[string]map[int32]string
+	id          string
+	state       int
+	wg          sync.WaitGroup
+	stopCh      chan struct{}
+	triggerCh   chan int
+	triggerOnce *sync.Once
+	owners      map[string]map[int32]string
 
 	config *Config
 	logger *logrus.Logger
@@ -62,7 +62,6 @@ func NewConsumerGroup(config *Config) (*ConsumerGroup, error) {
 		cg.id = genConsumerID()
 	}
 	cg.name = config.GroupID
-	cg.stopOnce = new(sync.Once)
 	cg.triggerCh = make(chan int)
 	cg.topicConsumers = make(map[string]*topicConsumer)
 	cg.onLoad = make([]func(), 0)
@@ -129,10 +128,6 @@ func (cg *ConsumerGroup) IsStopped() bool {
 	return cg.state == cgStopped
 }
 
-func (cg *ConsumerGroup) triggerRebalance() {
-	cg.triggerCh <- restartEvent
-}
-
 func (cg *ConsumerGroup) callRecover() {
 	if err := recover(); err != nil {
 		cg.logger.WithFields(logrus.Fields{
@@ -167,6 +162,7 @@ func (cg *ConsumerGroup) start() {
 CONSUME_TOPIC_LOOP:
 	for {
 		cg.logger.WithField("group", cg.name).Info("Consumer group started")
+		cg.triggerOnce = new(sync.Once)
 		cg.stopCh = make(chan struct{})
 
 		err := cg.watchRebalance()
@@ -191,6 +187,10 @@ CONSUME_TOPIC_LOOP:
 				defer cg.callRecover()
 				defer wg.Done()
 				tc.wg.Wait()
+				cg.logger.WithFields(logrus.Fields{
+					"group": tc.group,
+					"topic": tc.name,
+				}).Info("Stop the topic consumer")
 			}(consumer)
 		}
 		cg.state = cgStart
@@ -219,7 +219,11 @@ CONSUME_TOPIC_LOOP:
 }
 
 func (cg *ConsumerGroup) stop() {
-	cg.stopOnce.Do(func() { cg.triggerCh <- quitEvent })
+	cg.triggerOnce.Do(func() { cg.triggerCh <- quitEvent })
+}
+
+func (cg *ConsumerGroup) triggerRebalance() {
+	cg.triggerOnce.Do(func() { cg.triggerCh <- restartEvent })
 }
 
 func (cg *ConsumerGroup) getPartitionConsumer(topic string, partition int32, nextOffset int64) (sarama.PartitionConsumer, error) {

partition.go

@@ -43,7 +43,7 @@ func (pc *partitionConsumer) start() {
 			"topic":     pc.topic,
 			"partition": pc.partition,
 			"err":       err,
-		}).Error("Failed to clamin the partition and gave up")
+		}).Error("Failed to claim the partition and gave up")
 		goto ERROR
 	}
 	defer func() {
@@ -166,7 +166,7 @@ func (pc *partitionConsumer) claim() error {
 				"partition": pc.partition,
 				"retries":   i,
 				"err":       err,
-			}).Warn("Failed to claim the partiton with retries")
+			}).Warn("Failed to claim the partition with retries")
 		}
 		select {
 		case <-timer.C:

topic.go

@@ -37,10 +37,6 @@ func (tc *topicConsumer) start() {
 		"group": tc.group,
 		"topic": topic,
 	}).Info("Start the topic consumer")
-	defer cg.logger.WithFields(logrus.Fields{
-		"group": tc.group,
-		"topic": topic,
-	}).Info("Stop the topic consumer")
 
 	partitions, err := tc.assignPartitions()
 	if err != nil {