code: func (ref *Consumer) messageHandler(in <-chan *sarama.ConsumerMessage) {
  	ref.Debug("messageHandler started ...")
  	var prevValue []byte
  
  	for {
  		select {
  		case msg := <-in:
  			if msg == nil {
  				continue
  			}
  			consumerMsg := &ConsumerMessage{
  				Key:       msg.Key,
  				Value:     msg.Value,
  				PrevValue: prevValue,
  				Topic:     msg.Topic,
  				Partition: msg.Partition,
  				Offset:    msg.Offset,
  				Timestamp: msg.Timestamp,
  			}
  			// Store value as previous for the next iteration
  			prevValue = consumerMsg.Value
  			select {
  			case ref.Config.RecvMessageChan <- consumerMsg:
  			case <-time.After(1 * time.Second):
  				ref.Warn("Failed to deliver a message")
  			}
  		case <-ref.closeChannel:
  			ref.Debug("Canceling message handler")
  			return
  		}
  	}
  }
docs: // messageHandler processes each incoming message