code: func (ref *Consumer) StartConsumerManualHandlers(partitionConsumer sarama.PartitionConsumer) {
  	config := ref.Config
  	config.Logger.Info("Starting message handlers for new manual consumer ...")
  
  	// if required, start reading from the errors channel
  	if config.ProducerConfig().Consumer.Return.Errors {
  		go ref.manualErrorHandler(partitionConsumer.Errors())
  	}
  
  	// start the message handler
  	go ref.messageHandler(partitionConsumer.Messages())
  }
docs: // StartConsumerManualHandlers starts required handlers using sarama partition consumer. Used when partitioner set in config is
  // manual