From a933190fc72a98c701bb45cd09b747db7c6bcc13 Mon Sep 17 00:00:00 2001
From: Ti Chi Robot <ti-community-prow-bot@tidb.io>
Date: Wed, 22 Sep 2021 12:26:45 +0800
Subject: [PATCH] sink: fix codec json encoding with non binary string (#2764)
 (#2784)

---
 cdc/sink/codec/json.go      | 10 +++++++++-
 cdc/sink/codec/json_test.go | 22 ++++++++++++++++++++++
 2 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/cdc/sink/codec/json.go b/cdc/sink/codec/json.go
index d72e3ccf224..419f4a83334 100644
--- a/cdc/sink/codec/json.go
+++ b/cdc/sink/codec/json.go
@@ -64,7 +64,15 @@ func (c *column) FromSinkColumn(col *model.Column) {
 	}
 	switch col.Type {
 	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
-		str := string(col.Value.([]byte))
+		var str string
+		switch col.Value.(type) {
+		case []byte:
+			str = string(col.Value.([]byte))
+		case string:
+			str = col.Value.(string)
+		default:
+			log.Panic("invalid column value, please report a bug", zap.Any("col", col))
+		}
 		if c.Flag.IsBinary() {
 			str = strconv.Quote(str)
 			str = str[1 : len(str)-1]
diff --git a/cdc/sink/codec/json_test.go b/cdc/sink/codec/json_test.go
index f84afd16e80..f5cb63a8970 100644
--- a/cdc/sink/codec/json_test.go
+++ b/cdc/sink/codec/json_test.go
@@ -363,6 +363,28 @@ func (s *columnSuite) TestFormatCol(c *check.C) {
 	c.Assert(row2, check.DeepEquals, row)
 }
 
+func (s *columnSuite) TestNonBinaryStringCol(c *check.C) {
+	defer testleak.AfterTest(c)()
+	col := &model.Column{
+		Name:  "test",
+		Type:  mysql.TypeString,
+		Value: "value",
+	}
+	jsonCol := column{}
+	jsonCol.FromSinkColumn(col)
+	row := &mqMessageRow{Update: map[string]column{"test": jsonCol}}
+	rowEncode, err := row.Encode()
+	c.Assert(err, check.IsNil)
+	row2 := new(mqMessageRow)
+	err = row2.Decode(rowEncode)
+	c.Assert(err, check.IsNil)
+	c.Assert(row2, check.DeepEquals, row)
+	jsonCol2 := row2.Update["test"]
+	col2 := jsonCol2.ToSinkColumn("test")
+	col2.Value = string(col2.Value.([]byte))
+	c.Assert(col2, check.DeepEquals, col)
+}
+
 func (s *columnSuite) TestVarBinaryCol(c *check.C) {
 	defer testleak.AfterTest(c)()
 	col := &model.Column{