func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
    close(consumer.ready)
    consumer.messageChan = make(chan []byte, 1000)
    go handleLogMessage(consumer.messageChan)
    return nil
}
Setup is run at the beginning of a new session, before ConsumeClaim
package main
import "github.com/Shopify/sarama"
type Consumer struct {
    ready chan bool
    messageChan chan []byte
}
func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for message := range claim.Messages() {
        consumer.messageChan <- message.Value
    }
    return nil
}