code: func (s *Source) Record() (interface{}, error) {
  	if s.MaxMsgs > 0 {
  		s.numMsgs++
  		if s.numMsgs > s.MaxMsgs {
  			return nil, io.EOF
  		}
  	}
  	msg, ok := <-s.consumer.Messages()
  	if ok {
  		var ret interface{}
  		switch s.Type {
  		case "json":
  			parsed := make(map[string]interface{})
  			err := json.Unmarshal(msg.Value, &parsed)
  			if err != nil {
  				return nil, errors.Wrap(err, "unmarshaling json")
  			}
  			ret = parsed
  		case "raw":
  			ret = msg
  		default:
  			return nil, errors.Errorf("unsupported kafka message type: '%v'", s.Type)
  		}
  		s.consumer.MarkOffset(msg, "") // mark message as processed
  		return ret, nil
  	}
  	return nil, errors.New("messages channel closed")
  }
docs: // Record returns the value of the next kafka message.