我的书签
添加书签
移除书签基于 Avro 的 TiCDC 行数据 Checksum 校验
本文介绍如何使用 Golang 消费 TiCDC 发送到 Kafka、且由 Avro 协议编码的数据，以及如何基于单行数据 Checksum 功能进行数据校验。
本示例代码位于 avro-checksum-verification 目录下。
本文使用 kafka-go 实现一个简单的 Kafka Consumer 程序。该程序不断地从指定的 Topic 中读取数据、计算并校验 Checksum 值。
1. package main
3. import (
4. "context"
5. "encoding/binary"
6. "encoding/json"
7. "hash/crc32"
8. "io"
9. "math"
10. "net/http"
11. "strconv"
12. "strings"
14. "github.com/linkedin/goavro/v2"
15. "github.com/pingcap/log"
16. "github.com/pingcap/tidb/parser/mysql"
17. "github.com/pingcap/tidb/types"
18. "github.com/pingcap/tiflow/pkg/errors"
19. "github.com/segmentio/kafka-go"
20. "go.uber.org/zap"
21. )
23. const (
24. // confluent avro wire format, the first byte is always 0
25. // https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/index.html#wire-format
26. magicByte = uint8(0)
27. )
29. func main() {
30. var (
31. kafkaAddr = "127.0.0.1:9092"
32. schemaRegistryURL = "http://127.0.0.1:8081"
34. topic = "avro-checksum-test"
35. consumerGroupID = "avro-checksum-test"
36. )
38. consumer := kafka.NewReader(kafka.ReaderConfig{
39. Brokers: []string{kafkaAddr},
40. GroupID: consumerGroupID,
41. Topic: topic,
42. MaxBytes: 10e6, // 10MB
43. })
44. defer consumer.Close()
46. ctx := context.Background()
47. log.Info("start consuming ...", zap.String("kafka", kafkaAddr), zap.String("topic", topic), zap.String("groupID", consumerGroupID))
48. for {
49. // 1. 获取 kafka 消息
50. message, err := consumer.FetchMessage(ctx)
51. if err != nil {
52. log.Error("read kafka message failed", zap.Error(err))
53. }
55. value := message.Value
56. if len(value) == 0 {
57. log.Info("delete event does not have value, skip checksum verification", zap.String("topic", topic))
58. }
60. // 2. 对 value 进行解码，得到对应的 value map 和 schema map
61. valueMap, valueSchema, err := getValueMapAndSchema(value, schemaRegistryURL)
62. if err != nil {
63. log.Panic("decode kafka value failed", zap.String("topic", topic), zap.ByteString("value", value), zap.Error(err))
64. }
66. // 3. 使用上一步得到的 value map 和 schema map，计算并且校验 checksum
67. err = CalculateAndVerifyChecksum(valueMap, valueSchema)
68. if err != nil {
69. log.Panic("calculate checksum failed", zap.String("topic", topic), zap.ByteString("value", value), zap.Error(err))
70. }
72. // 4. 数据消费成功，提交 offset
73. if err := consumer.CommitMessages(ctx, message); err != nil {
74. log.Error("commit kafka message failed", zap.Error(err))
75. break
76. }
77. }
78. }
从上面的代码可以看出，getValueMapAndSchema() 和 CalculateAndVerifyChecksum() 是计算 Checksum 的关键步骤，下面分别介绍这两个函数的实现。
解码数据以及获取相应的 Schema
getValueMapAndSchema() 方法的主要作用是解码数据以及获取相应的 schema，二者均以 map[string]interface{} 类型返回。
1. // data is received kafka message's key or value, url is the schema registry url.
2. // return the decoded value and corresponding schema as map.
3. func getValueMapAndSchema(data []byte, url string) (map[string]interface{}, map[string]interface{}, error) {
4. schemaID, binary, err := extractSchemaIDAndBinaryData(data)
5. if err != nil {
6. return nil, nil, err
7. }
9. codec, err := GetSchema(url, schemaID)
10. if err != nil {
11. return nil, nil, err
12. }
14. native, _, err := codec.NativeFromBinary(binary)
15. if err != nil {
16. return nil, nil, err
17. }
19. result, ok := native.(map[string]interface{})
20. if !ok {
21. return nil, nil, errors.New("raw avro message is not a map")
22. }
24. schema := make(map[string]interface{})
25. if err := json.Unmarshal([]byte(codec.Schema()), &schema); err != nil {
26. return nil, nil, errors.Trace(err)
27. }
29. return result, schema, nil
30. }
32. // extractSchemaIDAndBinaryData
33. func extractSchemaIDAndBinaryData(data []byte) (int, []byte, error) {
34. if len(data) < 5 {
35. return 0, nil, errors.ErrAvroInvalidMessage.FastGenByArgs()
36. }
37. if data[0] != magicByte {
38. return 0, nil, errors.ErrAvroInvalidMessage.FastGenByArgs()
39. }
40. return int(binary.BigEndian.Uint32(data[1:5])), data[5:], nil
41. }
43. // GetSchema query the schema registry to fetch the schema by the schema id.
44. // return the goavro.Codec which can be used to encode and decode the data.
45. func GetSchema(url string, schemaID int) (*goavro.Codec, error) {
46. requestURI := url + "/schemas/ids/" + strconv.Itoa(schemaID)
48. req, err := http.NewRequest("GET", requestURI, nil)
49. if err != nil {
50. log.Error("Cannot create the request to look up the schema", zap.Error(err))
51. return nil, errors.WrapError(errors.ErrAvroSchemaAPIError, err)
52. }
53. req.Header.Add(
54. "Accept",
55. "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, "+
56. "application/json",
57. )
59. httpClient := &http.Client{}
60. resp, err := httpClient.Do(req)
61. if err != nil {
62. return nil, err
63. }
64. defer resp.Body.Close()
66. body, err := io.ReadAll(resp.Body)
67. if err != nil {
68. log.Error("Cannot parse the lookup schema response", zap.Error(err))
69. return nil, errors.WrapError(errors.ErrAvroSchemaAPIError, err)
70. }
72. if resp.StatusCode == 404 {
73. log.Warn("Specified schema not found in Registry", zap.String("requestURI", requestURI), zap.Int("schemaID", schemaID))
74. return nil, errors.ErrAvroSchemaAPIError.GenWithStackByArgs("Schema not found in Registry")
75. }
77. if resp.StatusCode != 200 {
78. log.Error("Failed to query schema from the Registry, HTTP error",
79. zap.Int("status", resp.StatusCode), zap.String("uri", requestURI), zap.ByteString("responseBody", body))
80. return nil, errors.ErrAvroSchemaAPIError.GenWithStack("Failed to query schema from the Registry, HTTP error")
81. }
83. var jsonResp lookupResponse
84. err = json.Unmarshal(body, &jsonResp)
85. if err != nil {
86. log.Error("Failed to parse result from Registry", zap.Error(err))
87. return nil, errors.WrapError(errors.ErrAvroSchemaAPIError, err)
88. }
90. codec, err := goavro.NewCodec(jsonResp.Schema)
91. if err != nil {
92. return nil, errors.WrapError(errors.ErrAvroSchemaAPIError, err)
93. }
94. return codec, nil
95. }
97. type lookupResponse struct {
98. Name string `json:"name"`
99. SchemaID int `json:"id"`
100. Schema string `json:"schema"`
101. }
计算并校验 Checksum
上一步获取的 valueMap 和 valueSchema 包含了所有用于 Checksum 计算和校验的元素。
在消费端计算和校验 Checksum 的过程包含以下几个步骤：
- 获取期望的 Checksum 值。
- 遍历每一列，根据列的数据值和对应的 MySQL Type，生成字节切片，不断更新 Checksum 值。
- 将上一步计算得到的 Checksum 和从收到的消息里提取出来的 Checksum 做比较。如果不一致，则说明 Checksum 校验失败，数据可能发生损坏。
示例代码如下：
1. func CalculateAndVerifyChecksum(valueMap, valueSchema map[string]interface{}) error {
2. // fields 存放有数据变更事件的每一个列的类型信息，按照每一列的 ID 排序，该顺序和 Checksum 计算顺序相同
3. fields, ok := valueSchema["fields"].([]interface{})
4. if !ok {
5. return errors.New("schema fields should be a map")
6. }
8. // 1. 从 valueMap 里面查找期望的 checksum 值，它被编码成 string 类型
9. // 如果找不到，说明 TiCDC 发送该条数据时，还没有开启 checksum 功能，直接返回即可
10. o, ok := valueMap["_tidb_row_level_checksum"]
11. if !ok {
12. return nil
13. }
14. expected := o.(string)
15. if expected == "" {
16. return nil
17. }
19. // expectedChecksum 即是从 TiCDC 传递而来的期望的 checksum 值
20. expectedChecksum, err := strconv.ParseUint(expected, 10, 64)
21. if err != nil {
22. return errors.Trace(err)
23. }
25. // 2. 遍历每一个 field，计算 checksum 值
26. var actualChecksum uint32
27. // buf 用来存储每次更新 checksum 时使用的字节切片
28. buf := make([]byte, 0)
29. for _, item := range fields {
30. field, ok := item.(map[string]interface{})
31. if !ok {
32. return errors.New("schema field should be a map")
33. }
35. // `tidbOp` 及之后的列不参与到 checksum 计算中，因为它们是一些用于辅助数据消费的列，并非真实的 TiDB 列数据
36. colName := field["name"].(string)
37. if colName == "_tidb_op" {
38. break
39. }
41. // holder 存放有列类型信息
42. var holder map[string]interface{}
43. switch ty := field["type"].(type) {
44. case []interface{}:
45. for _, item := range ty {
46. if m, ok := item.(map[string]interface{}); ok {
47. holder = m["connect.parameters"].(map[string]interface{})
48. break
49. }
50. }
51. case map[string]interface{}:
52. holder = ty["connect.parameters"].(map[string]interface{})
53. default:
54. log.Panic("type info is anything else", zap.Any("typeInfo", field["type"]))
55. }
56. tidbType := holder["tidb_type"].(string)
58. mysqlType := mysqlTypeFromTiDBType(tidbType)
60. // 根据每一列的名字，从解码之后的 value map 里拿到该列的值
61. value, ok := valueMap[colName]
62. if !ok {
63. return errors.New("value not found")
64. }
65. value, err := getColumnValue(value, holder, mysqlType)
66. if err != nil {
67. return errors.Trace(err)
68. }
70. if len(buf) > 0 {
71. buf = buf[:0]
72. }
74. // 根据每一列的 value 和 mysqlType，生成用于更新 checksum 的字节切片，然后更新 checksum
75. buf, err = buildChecksumBytes(buf, value, mysqlType)
76. if err != nil {
77. return errors.Trace(err)
78. }
79. actualChecksum = crc32.Update(actualChecksum, crc32.IEEETable, buf)
80. }
82. if uint64(actualChecksum) != expectedChecksum {
83. log.Error("checksum mismatch",
84. zap.Uint64("expected", expectedChecksum),
85. zap.Uint64("actual", uint64(actualChecksum)))
86. return errors.New("checksum mismatch")
87. }
89. log.Info("checksum verified", zap.Uint64("checksum", uint64(actualChecksum)))
90. return nil
91. }
93. func mysqlTypeFromTiDBType(tidbType string) byte {
94. var result byte
95. switch tidbType {
96. case "INT", "INT UNSIGNED":
97. result = mysql.TypeLong
98. case "BIGINT", "BIGINT UNSIGNED":
99. result = mysql.TypeLonglong
100. case "FLOAT":
101. result = mysql.TypeFloat
102. case "DOUBLE":
103. result = mysql.TypeDouble
104. case "BIT":
105. result = mysql.TypeBit
106. case "DECIMAL":
107. result = mysql.TypeNewDecimal
108. case "TEXT":
109. result = mysql.TypeVarchar
110. case "BLOB":
111. result = mysql.TypeLongBlob
112. case "ENUM":
113. result = mysql.TypeEnum
114. case "SET":
115. result = mysql.TypeSet
116. case "JSON":
117. result = mysql.TypeJSON
118. case "DATE":
119. result = mysql.TypeDate
120. case "DATETIME":
121. result = mysql.TypeDatetime
122. case "TIMESTAMP":
123. result = mysql.TypeTimestamp
124. case "TIME":
125. result = mysql.TypeDuration
126. case "YEAR":
127. result = mysql.TypeYear
128. default:
129. log.Panic("this should not happen, unknown TiDB type", zap.String("type", tidbType))
130. }
131. return result
132. }
134. // value 是一个 interface 类型的值，需要根据 holder 提供的类型信息，做一次转换处理
135. func getColumnValue(value interface{}, holder map[string]interface{}, mysqlType byte) (interface{}, error) {
136. switch t := value.(type) {
137. // nullable 的列，其值被编码成一个 map，只有一个键值对，键是类型，值是真实的值，此处只关心真实的值
138. case map[string]interface{}:
139. for _, v := range t {
140. value = v
141. }
142. }
144. switch mysqlType {
145. case mysql.TypeEnum:
146. // Enum 被编码成了 string，此处转换为对应于 Enum 定义的 int 值
147. allowed := strings.Split(holder["allowed"].(string), ",")
148. switch t := value.(type) {
149. case string:
150. enum, err := types.ParseEnum(allowed, t, "")
151. if err != nil {
152. return nil, errors.Trace(err)
153. }
154. value = enum.Value
155. case nil:
156. value = nil
157. }
158. case mysql.TypeSet:
159. // Set 被编码成了 string，根据 set 定义的顺序，转换为对应的 int 值
160. elems := strings.Split(holder["allowed"].(string), ",")
161. switch t := value.(type) {
162. case string:
163. s, err := types.ParseSet(elems, t, "")
164. if err != nil {
165. return nil, errors.Trace(err)
166. }
167. value = s.Value
168. case nil:
169. value = nil
170. }
171. }
172. return value, nil
173. }
175. // buildChecksumBytes 生成用于更新 checksum 的字节切片, 参考 https://github.com/pingcap/tidb/blob/e3417913f58cdd5a136259b902bf177eaf3aa637/util/rowcodec/common.go#L308
176. func buildChecksumBytes(buf []byte, value interface{}, mysqlType byte) ([]byte, error) {
177. if value == nil {
178. return buf, nil
179. }
181. switch mysqlType {
182. // TypeTiny, TypeShort, TypeInt32 被编码成 int32
183. // TypeLong 被编码成 int32 if signed, else int64
184. // TypeLongLong，如果是 signed，被编码成 int64，否则被编码成 uint64,
185. // 开启 checksum 功能，bigintUnsignedHandlingMode 必须设置为 string，被编码成 string.
186. case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeLonglong, mysql.TypeInt24, mysql.TypeYear:
187. switch a := value.(type) {
188. case int32:
189. buf = binary.LittleEndian.AppendUint64(buf, uint64(a))
190. case uint32:
191. buf = binary.LittleEndian.AppendUint64(buf, uint64(a))
192. case int64:
193. buf = binary.LittleEndian.AppendUint64(buf, uint64(a))
194. case uint64:
195. buf = binary.LittleEndian.AppendUint64(buf, a)
196. case string:
197. v, err := strconv.ParseUint(a, 10, 64)
198. if err != nil {
199. return nil, errors.Trace(err)
200. }
201. buf = binary.LittleEndian.AppendUint64(buf, v)
202. default:
203. log.Panic("unknown golang type for the integral value",
204. zap.Any("value", value), zap.Any("mysqlType", mysqlType))
205. }
206. // Float 类型编码为 float32，Double 编码为 float64
207. case mysql.TypeFloat, mysql.TypeDouble:
208. var v float64
209. switch a := value.(type) {
210. case float32:
211. v = float64(a)
212. case float64:
213. v = a
214. }
215. if math.IsInf(v, 0) || math.IsNaN(v) {
216. v = 0
217. }
218. buf = binary.LittleEndian.AppendUint64(buf, math.Float64bits(v))
219. // getColumnValue 将 Enum 和 Set 转换为了 uint64 类型
220. case mysql.TypeEnum, mysql.TypeSet:
221. buf = binary.LittleEndian.AppendUint64(buf, value.(uint64))
222. case mysql.TypeBit:
223. // bit 类型编码为 []bytes，需要进一步转换为 uint64
224. v, err := binaryLiteralToInt(value.([]byte))
225. if err != nil {
226. return nil, errors.Trace(err)
227. }
228. buf = binary.LittleEndian.AppendUint64(buf, v)
229. // 非二进制类型时，编码成 string， 反之则为 []byte
230. case mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeString, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob, mysql.TypeBlob:
231. switch a := value.(type) {
232. case string:
233. buf = appendLengthValue(buf, []byte(a))
234. case []byte:
235. buf = appendLengthValue(buf, a)
236. default:
237. log.Panic("unknown golang type for the string value",
238. zap.Any("value", value), zap.Any("mysqlType", mysqlType))
239. }
240. case mysql.TypeTimestamp, mysql.TypeDatetime, mysql.TypeDate, mysql.TypeDuration, mysql.TypeNewDate:
241. v := value.(string)
242. buf = appendLengthValue(buf, []byte(v))
243. // 开启 checksum 功能时，decimalHandlingMode 必须设置为 string
244. case mysql.TypeNewDecimal:
245. buf = appendLengthValue(buf, []byte(value.(string)))
246. case mysql.TypeJSON:
247. buf = appendLengthValue(buf, []byte(value.(string)))
248. // Null 和 Geometry 不参与到 checksum 计算
249. case mysql.TypeNull, mysql.TypeGeometry:
250. // do nothing
251. default:
252. return buf, errors.New("invalid type for the checksum calculation")
253. }
254. return buf, nil
255. }
257. func appendLengthValue(buf []byte, val []byte) []byte {
258. buf = binary.LittleEndian.AppendUint32(buf, uint32(len(val)))
259. buf = append(buf, val...)
260. return buf
261. }
263. // 将 []byte 转换为 uint64，参考 https://github.com/pingcap/tidb/blob/e3417913f58cdd5a136259b902bf177eaf3aa637/types/binary_literal.go#L105
264. func binaryLiteralToInt(bytes []byte) (uint64, error) {
265. bytes = trimLeadingZeroBytes(bytes)
266. length := len(bytes)
268. if length > 8 {
269. log.Error("invalid bit value found", zap.ByteString("value", bytes))
270. return math.MaxUint64, errors.New("invalid bit value")
271. }
273. if length == 0 {
274. return 0, nil
275. }
277. val := uint64(bytes[0])
278. for i := 1; i < length; i++ {
279. val = (val << 8) | uint64(bytes[i])
280. }
281. return val, nil
282. }
284. func trimLeadingZeroBytes(bytes []byte) []byte {
285. if len(bytes) == 0 {
286. return bytes
287. }
288. pos, posMax := 0, len(bytes)-1
289. for ; pos < posMax; pos++ {
290. if bytes[pos] != 0 {
291. break
292. }
293. }
294. return bytes[pos:]
295. }
