From c4e93218ba2e32b70fa62284f5331dc4f42f4359 Mon Sep 17 00:00:00 2001
From: Ti Chi Robot <ti-community-prow-bot@tidb.io>
Date: Sat, 11 Sep 2021 00:50:39 +0800
Subject: [PATCH] sink: fix codec json encoding with non binary string (#2764)
 (#2781)

---
 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 09bdf7cb429..edacbd7878f 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 e6d8542499c..98d56739a8e 100644
--- a/cdc/sink/codec/json_test.go
+++ b/cdc/sink/codec/json_test.go
@@ -387,6 +387,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{