From 331dd63c5ff4b711792b0c50a7ec9bd47346f70e Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Tue, 14 May 2019 17:31:51 +0800
Subject: [PATCH] =?UTF-8?q?Message=E8=BD=ACDml=E5=AF=B9null=E5=80=BC?=
 =?UTF-8?q?=E7=9A=84=E5=88=A4=E6=96=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../client/adapter/support/MessageUtil.java   | 28 +++++++++++--------
 1 file changed, 16 insertions(+), 12 deletions(-)

diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java
index 1e24d9ad..004b7087 100644
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java
+++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java
@@ -22,7 +22,7 @@ public class MessageUtil {
         List<Dml> dmls = new ArrayList<Dml>(entries.size());
         for (CanalEntry.Entry entry : entries) {
             if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
-                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
+                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                 continue;
             }
 
@@ -31,7 +31,7 @@ public class MessageUtil {
                 rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
             } catch (Exception e) {
                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
-                    e);
+                        e);
             }
 
             CanalEntry.EventType eventType = rowChange.getEventType();
@@ -57,7 +57,7 @@ public class MessageUtil {
                 int i = 0;
                 for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                     if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE
-                        && eventType != CanalEntry.EventType.DELETE) {
+                            && eventType != CanalEntry.EventType.DELETE) {
                         continue;
                     }
 
@@ -76,11 +76,15 @@ public class MessageUtil {
                                 dml.getPkNames().add(column.getName());
                             }
                         }
-                        row.put(column.getName(),
-                            JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
-                                column.getValue(),
-                                column.getSqlType(),
-                                column.getMysqlType()));
+                        if (column.getIsNull()) {
+                            row.put(column.getName(), null);
+                        } else {
+                            row.put(column.getName(),
+                                    JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(),
+                                            column.getValue(),
+                                            column.getSqlType(),
+                                            column.getMysqlType()));
+                        }
                         // 获取update为true的字段
                         if (column.getUpdated()) {
                             updateSet.add(column.getName());
@@ -95,10 +99,10 @@ public class MessageUtil {
                         for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                             if (updateSet.contains(column.getName())) {
                                 rowOld.put(column.getName(),
-                                    JdbcTypeUtil.typeConvert(dml.getTable(),column.getName(),
-                                        column.getValue(),
-                                        column.getSqlType(),
-                                        column.getMysqlType()));
+                                        JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(),
+                                                column.getValue(),
+                                                column.getSqlType(),
+                                                column.getMysqlType()));
                             }
                         }
                         // update操作将记录修改前的值