code: func (s *Source) Open() error {
  	// init (custom) config, enable errors and notifications
  	sarama.Logger = log.New(ioutil.Discard, "", 0)
  	config := cluster.NewConfig()
  	config.Config.Version = sarama.V0_10_0_0
  	config.Consumer.Return.Errors = true
  	config.Consumer.Offsets.Initial = sarama.OffsetOldest
  	config.Group.Return.Notifications = true
  
  	var err error
  	s.consumer, err = cluster.NewConsumer(s.Hosts, s.Group, s.Topics, config)
  	if err != nil {
  		return errors.Wrap(err, "getting new consumer")
  	}
  	s.messages = s.consumer.Messages()
  
  	// consume errors
  	go func() {
  		for err := range s.consumer.Errors() {
  			log.Printf("Error: %s\n", err.Error())
  		}
  	}()
  
  	// consume notifications
  	go func() {
  		for ntf := range s.consumer.Notifications() {
  			log.Printf("Rebalanced: %+v\n", ntf)
  		}
  	}()
  	return nil
  }
docs: // Open initializes the kafka source.