code: func (mux *Multiplexer) manualConsumer(consumer *client.Consumer) {
  	mux.Debug("Generic Consumer started")
  	for {
  		select {
  		case <-consumer.GetCloseChannel():
  			mux.Debug("Closing Consumer")
  			return
  		case msg := <-consumer.Config.RecvMessageChan:
  			mux.Debug("Kafka message received")
  			// 'later-stage' Consumer does not consume 'hash' messages, none of them is marked
  			mux.propagateMessage(msg)
  		case err := <-consumer.Config.RecvErrorChan:
  			mux.Error("Received partitionConsumer error ", err)
  		}
  	}
  }
docs: // manualConsumer takes a consumer (even a post-init created) and handles incoming messages for them.