func convertRowToAvroFormat(tp int, row *[]interface{}, s *types.TableSchema, seqNo uint64, r *goavro.Record, filter []int) error {
	err := r.Set("ref_key", int64(seqNo))
	if err != nil {
		return err
	}
	switch tp {
	case types.Insert:
		err = fillAvroKey(r, row, s)
		if err != nil {
			return err
		}
		err = fillAvroFields(r, row, s, filter)
		if err != nil {
			return err
		}
		err = r.Set("is_deleted", false)
		if err != nil {
			return err
		}
	case types.Delete:
		err = fillAvroKey(r, row, s)
		if err != nil {
			return err
		}
		err = r.Set("is_deleted", true)
		if err != nil {
			return err
		}
	default:
		return fmt.Errorf("unknown event type: %v", tp)
	}
	return nil
}
convertRowToAvroFormat uses fillAvroKey and fillAvroFields to convert
the complete Avro record from row
package encoder
import (
	"bytes"
	"fmt"
	"strings"
	"time"
	"github.com/linkedin/goavro"
	"github.com/uber/storagetapper/log"
	"github.com/uber/storagetapper/state"
	"github.com/uber/storagetapper/types"
)
const numMetadataFields = 3
func init() {
	registerPlugin("avro", initAvroEncoder)
}
type avroEncoder struct {
	Service   string
	DB        string
	Table     string
	Input     string
	Output    string
	Version   int
	codec     goavro.Codec
	setter    *goavro.RecordSetter
	inSchema  *types.TableSchema
	filter    []int
	outSchema *types.AvroSchema
}
func initAvroEncoder(service, db, table, input string, output string, version int) (Encoder, error) {
	return &avroEncoder{Service: service, DB: db, Table: table, Input: input, Output: output, Version: version}, nil
}
func (e *avroEncoder) Type() string {
	return "avro"
}
func (e *avroEncoder) Schema() *types.TableSchema {
	return e.inSchema
}
func (e *avroEncoder) EncodeSchema(seqno uint64) ([]byte, error) {
	return nil, nil
}
func (e *avroEncoder) Row(tp int, row *[]interface{}, seqno uint64, _ time.Time) ([]byte, error) {
	r, err := goavro.NewRecord(*e.setter)
	if err != nil {
		return nil, err
	}
	err = convertRowToAvroFormat(tp, row, e.inSchema, seqno, r, e.filter)
	if err != nil {
		return nil, err
	}
	return encodeAvroRecord(e.codec, r)
}
func (e *avroEncoder) CommonFormat(cf *types.CommonFormatEvent) ([]byte, error) {
	if cf.Type == "schema" {
		err := e.UpdateCodec()
		return nil, err
	}
	r, err := goavro.NewRecord(*e.setter)
	if err != nil {
		return nil, err
	}
	err = convertCommonFormatToAvroRecord(*e.setter, cf, r, e.filter)
	if err != nil {
		return nil, err
	}
	return encodeAvroRecord(e.codec, r)
}
func convertCommonFormatToAvroRecord(rs goavro.RecordSetter, cfEvent *types.CommonFormatEvent, rec *goavro.Record, filter []int) error {
	err := rec.Set("row_key", []byte(GetCommonFormatKey(cfEvent)))  
	if err != nil {
		return err
	}
	err = rec.Set("ref_key", int64(cfEvent.SeqNo))
	if err != nil {
		return err
	}
	err = rec.Set("is_deleted", strings.EqualFold(cfEvent.Type, "delete"))
	if err != nil {
		return err
	}
	if cfEvent.Fields == nil {
		return nil
	}
	for i, j := 0, 0; i < len(*cfEvent.Fields); i++ {
		if filteredField(filter, i, &j) {
			continue
		}
		field := (*cfEvent.Fields)[i]
		switch r := field.Value.(type) {
		case float64:
			s, err := rec.GetFieldSchema(field.Name)
			if err != nil {
				return err
			}
			for _, t := range s.(map[string]interface{})["type"].([]interface{}) {
				switch t.(string) {
				case "int":
					field.Value = int32(r)
				case "long":
					field.Value = int64(r)
				}
			}
		case time.Time:
			if !r.Equal(time.Time{}) {
				field.Value = r.UnixNano() / 1000000
			} else {
				field.Value = nil
			}
		}
		err = rec.Set(field.Name, field.Value)
		if err != nil {
			return err
		}
	}
	return nil
}
func (e *avroEncoder) UpdateCodec() error {
	var err error
	log.Debugf("Schema codec (%v) updating", e.Type())
	e.inSchema, err = state.GetSchema(e.Service, e.DB, e.Table, e.Input, e.Output, e.Version)
	if log.E(err) {
		return err
	}
	n, err := GetOutputSchemaName(e.Service, e.DB, e.Table, e.Input, e.Output, e.Version)
	if log.E(err) {
		return err
	}
	e.outSchema, err = GetLatestSchema("production", n, "avro")
	if log.E(err) {
		return err
	}
	if len(e.inSchema.Columns)-(len(e.outSchema.Fields)-numMetadataFields) < 0 {
		err = fmt.Errorf("input schema has less fields than output schema")
		log.E(err)
		return err
	}
	e.codec, e.setter, err = SchemaCodecHelper(e.outSchema)
	if log.E(err) {
		return err
	}
	e.filter = prepareFilter(e.inSchema, e.outSchema, numMetadataFields)
	log.Debugf("Schema codec (%v) updated", e.Type())
	return err
}
func encodeAvroRecord(codec goavro.Codec, r *goavro.Record) ([]byte, error) {
	w := new(bytes.Buffer)
	err := codec.Encode(w, r)
	if err != nil {
		return nil, err
	}
	return w.Bytes(), nil
}
func fillAvroKey(e *goavro.Record, row *[]interface{}, s *types.TableSchema) error {
	var rowKey string
	for i := 0; i < len(s.Columns); i++ {
		if s.Columns[i].Key == "PRI" {
			k := fmt.Sprintf("%v", (*row)[i])
			rowKey += fmt.Sprintf("%v%v", len(k), k)
		}
	}
	return e.Set("row_key", []byte(rowKey))
}
func convertTime(b string, dtype string) (interface{}, error) {
	var v interface{}
	if dtype == "timestamp" || dtype == "datetime" {
		if strings.HasPrefix(b, "0000-00-00 00:00:00") {
			return nil, nil
		}
		t, err := time.Parse(time.RFC3339Nano, b)
		if err != nil {
			return nil, err
		}
		v = t.UnixNano() / 1000000
	} else if dtype == "binary" || dtype == "varbinary" {
		v = []byte(b)
	} else {
		v = b
	}
	return v, nil
}
func convertText(b []byte, d string) interface{} {
	if d == "text" || d == "tinytext" || d == "mediumtext" || d == "longtext" || d == "json" {
		return string(b)
	}
	return b
}
func fixAvroFieldType(i interface{}, dtype string, ftype string) (interface{}, error) {
	var err error
	var v interface{}
	switch b := i.(type) {
	case int8:
		if ftype == types.MySQLBoolean {
			v = false
			if b != 0 {
				v = true
			}
		} else {
			v = int32(b)
		}
	case uint8:
		v = int32(b)
	case int16:
		v = int32(b)
	case uint16:
		v = int32(b)
	case time.Time:
		v = b.UnixNano() / 1000000  
	case string:
		v, err = convertTime(b, dtype)
		if err != nil {
			return nil, err
		}
	case []byte:
		v = convertText(b, dtype)
	default:
		v = b
	}
	return v, nil
}
func fillAvroFields(r *goavro.Record, row *[]interface{}, s *types.TableSchema, filter []int) error {
	for i, j := 0, 0; i < len(s.Columns); i++ {
		if filteredField(filter, i, &j) {
			continue
		}
		v, err := fixAvroFieldType((*row)[i], s.Columns[i].DataType, s.Columns[i].Type)
		if err != nil {
			return err
		}
		err = r.Set(s.Columns[i].Name, v)
		if err != nil {
			return err
		}
	}
	return nil
}