From dbd87b9a019e6c975fbdf2254a47b70f139f8201 Mon Sep 17 00:00:00 2001
From: Rinka <jiajidi@163.com>
Date: Thu, 21 Apr 2022 14:25:00 +0800
Subject: [PATCH] [hotfix][tidb] Fix RowKind.INSERT judgement condition (#1104)

Co-authored-by: rinka <gzlinjia@corp.netease.com>
---
 .../tidb/table/RowDataTiKVChangeEventDeserializationSchema.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-connector-tidb-cdc/src/main/java/com/ververica/cdc/connectors/tidb/table/RowDataTiKVChangeEventDeserializationSchema.java b/flink-connector-tidb-cdc/src/main/java/com/ververica/cdc/connectors/tidb/table/RowDataTiKVChangeEventDeserializationSchema.java
index 7e8e01540..ac6914edb 100644
--- a/flink-connector-tidb-cdc/src/main/java/com/ververica/cdc/connectors/tidb/table/RowDataTiKVChangeEventDeserializationSchema.java
+++ b/flink-connector-tidb-cdc/src/main/java/com/ververica/cdc/connectors/tidb/table/RowDataTiKVChangeEventDeserializationSchema.java
@@ -81,7 +81,7 @@ public class RowDataTiKVChangeEventDeserializationSchema
                                     row.getValue().toByteArray(),
                                     RowKey.decode(row.getKey().toByteArray()).getHandle(),
                                     tableInfo);
-                    if (row.getOldValue() == null) {
+                    if (row.getOldValue() == null || row.getOldValue().isEmpty()) {
                         RowData rowDataUpdateBefore =
                                 (RowData) physicalConverter.convert(tikvValues, tableInfo, null);
                         rowDataUpdateBefore.setRowKind(RowKind.INSERT);