code: func (s *ConfluentSource) Record() (interface{}, error) {
  	rec, err := s.Source.Record()
  	if err != nil {
  		return rec, err
  	}
  	msg, ok := rec.(*sarama.ConsumerMessage)
  	if !ok {
  		return rec, errors.Errorf("record is not a raw kafka record, but a %T", rec)
  	}
  	val := msg.Value
  	return s.decodeAvroValueWithSchemaRegistry(val)
  }
docs: // Record returns the next value from kafka.