Merge pull request #21 from meitu/Fix/sarama-nil-message

Fix/sarama nil message

config.go

@@ -44,6 +44,7 @@ func NewConfig() *Config {
 	config.OffsetAutoReset = sarama.OffsetNewest
 	config.ClaimPartitionRetryTimes = 10
 	config.ClaimPartitionRetryInterval = 3 * time.Second
+	config.SaramaConfig.Consumer.Return.Errors = true
 	return config
 }
 

example/example.go

@@ -37,7 +37,8 @@ func main() {
 	conf := consumergroup.NewConfig()
 	conf.ZkList = []string{"127.0.0.1:2181"}
 	conf.ZkSessionTimeout = 6 * time.Second
-	conf.TopicList = []string{"test"}
+	topic := "test"
+	conf.TopicList = []string{topic}
 	conf.GroupID = "go-test-group-id"
 
 	cg, err := consumergroup.NewConsumerGroup(conf)
@@ -56,16 +57,16 @@ func main() {
 
 	// Retrieve the error and log
 	go func() {
-		if topicErrChan, ok := cg.GetErrors("test"); ok {
+		if topicErrChan, ok := cg.GetErrors(topic); ok {
 			for err := range topicErrChan {
 				if err != nil {
-					fmt.Println("toipic %s got err, %s", err)
+					fmt.Printf("Toipic %s got err, %s\n", topic, err)
 				}
 			}
 		}
 	}()
 
-	if msgChan, ok := cg.GetMessages("test"); ok {
+	if msgChan, ok := cg.GetMessages(topic); ok {
 		for message := range msgChan {
 			fmt.Println(string(message.Value), message.Offset)
 			time.Sleep(500 * time.Millisecond)

partition.go

@@ -160,6 +160,11 @@ PARTITION_CONSUMER_LOOP:
 		case err := <-pc.consumer.Errors():
 			errorChan <- err
 		case message := <-pc.consumer.Messages():
+			if message == nil {
+				cg.logger.Errorf("Sarama partition consumer encounter error, the consumer would be exited")
+				close(cg.stopper)
+				break PARTITION_CONSUMER_LOOP
+			}
 			select {
 			case messageChan <- message:
 				pc.offset = message.Offset + 1