CDC avro sink panic with unsigned tinyint

[dveeden]

Bug Report

1. What did you do? If possible, provide a recipe for reproducing the error.

Prepare TiDB nightly with TiCDC master branch and Kafka with registry and avro support

tiup playground nightly --monitor=false # v5.0.0-nightly-20210823
./bin/cdc server
curl --silent --output docker-compose.yml https://raw.githubusercontent.com/confluentinc/cp-all-in-one/6.2.0-post/cp-all-in-one-community/docker-compose.yml
sudo docker-compose up -d
./bin/cdc cli changefeed create --no-confirm --changefeed-id="simple-replication-task" --sort-engine="unified" --sink-uri="kafka://127.0.0.1:9092/cdc-test?protocol=avro&kafka-version=2.8.0" --log-level debug --opts registry="http://127.0.0.1:8081"

CREATE TABLE t3 (id bigint unsigned auto_increment primary key, c1 tinyint(3) unsigned not null default '0');
insert into t3(c1) values(1);

2. What did you expect to see?

Replication from TiDB to Kafka

3. What did you see instead?

panic: interface conversion: interface {} is uint64, not int64

goroutine 3461 [running]:
github.com/pingcap/ticdc/cdc/sink/codec.columnToAvroNativeData(0xc0009fa3c0, 0xc000199260, 0xc002197350, 0x2, 0xc001670988, 0xd, 0x0, 0x0)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:487 +0x1893
github.com/pingcap/ticdc/cdc/sink/codec.rowToAvroNativeData(0xc001edf150, 0x2, 0x2, 0xc000199260, 0x4, 0xc002197328, 0x2, 0x38)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:272 +0xa8
github.com/pingcap/ticdc/cdc/sink/codec.avroEncode(0xc0009fa3f0, 0xc0015bc6e0, 0x5ee0d6c15ec0001, 0xc001edf150, 0x2, 0x2, 0xc000199260, 0x7f511de83d18, 0x48, 0x50)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:195 +0x191
github.com/pingcap/ticdc/cdc/sink/codec.(*AvroEventBatchEncoder).AppendRowChangedEvent(0xc002280030, 0xc0025dea80, 0x0, 0x0, 0x3)
	github.com/pingcap/ticdc/cdc/sink/codec/avro.go:92 +0x830
github.com/pingcap/ticdc/cdc/sink.(*mqSink).runWorker(0xc002350cf0, 0x33e1d30, 0xc00222d600, 0xc000000003, 0x0, 0x0)
	github.com/pingcap/ticdc/cdc/sink/mq.go:357 +0x3a9
github.com/pingcap/ticdc/cdc/sink.(*mqSink).run.func1(0x0, 0x2)
	github.com/pingcap/ticdc/cdc/sink/mq.go:287 +0x46
golang.org/x/sync/errgroup.(*Group).Go.func1(0xc002280000, 0xc001f22060)
	golang.org/x/sync@v0.0.0-20210220032951-036812b2e83c/errgroup/errgroup.go:57 +0x59
created by golang.org/x/sync/errgroup.(*Group).Go
	golang.org/x/sync@v0.0.0-20210220032951-036812b2e83c/errgroup/errgroup.go:54 +0x66

4. Versions of the cluster

   • Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

     5.7.25-TiDB-v5.2.0-alpha-714-g48e12ae07
     

   • TiCDC version (execute cdc version):

     Release Version: v5.2.0-master
     Git Commit Hash: 142ae26b291df8ca8ce8cb8286b2d92f87a261d4
     Git Branch: master
     UTC Build Time: 2021-08-26 14:57:17
     Go Version: go version go1.16.6 linux/amd64
     Failpoint Build: false
     

This also happens on release versions, not just on master/nightly.

[dveeden]

With this it no longer panics, but I didn't verify if it was working correctly

diff --git a/cdc/sink/codec/avro.go b/cdc/sink/codec/avro.go
index e72264e..f8820ec 100644
--- a/cdc/sink/codec/avro.go
+++ b/cdc/sink/codec/avro.go
@@ -484,7 +484,7 @@ func columnToAvroNativeData(col *model.Column, tz *time.Location) (interface{},
        case mysql.TypeBit:
                return handleUnsignedInt64()
        case mysql.TypeTiny, mysql.TypeShort, mysql.TypeInt24:
-               return int32(col.Value.(int64)), "int", nil
+               return handleUnsignedInt64()
        case mysql.TypeLong:
                if col.Flag.IsUnsigned() {
                        return int64(col.Value.(uint64)), "long", nil

[dveeden]

Or maybe:

diff --git a/cdc/sink/codec/avro.go b/cdc/sink/codec/avro.go
index e72264e..1d64529 100644
--- a/cdc/sink/codec/avro.go
+++ b/cdc/sink/codec/avro.go
@@ -484,6 +484,9 @@ func columnToAvroNativeData(col *model.Column, tz *time.Location) (interface{},
        case mysql.TypeBit:
                return handleUnsignedInt64()
        case mysql.TypeTiny, mysql.TypeShort, mysql.TypeInt24:
+               if col.Flag.IsUnsigned() {
+                       return int32(col.Value.(uint64)), "int", nil
+               }
                return int32(col.Value.(int64)), "int", nil
        case mysql.TypeLong:
                if col.Flag.IsUnsigned() {