From aeecf87490eee5dc297dfea7d3b5f25c660d8a80 Mon Sep 17 00:00:00 2001
From: machey <rewerma@163.com>
Date: Tue, 11 Dec 2018 00:09:12 +0800
Subject: [PATCH] =?UTF-8?q?=E5=8E=BB=E6=8E=89=E5=AF=B9before=E7=9A=84?=
 =?UTF-8?q?=E5=88=A4=E6=96=AD=EF=BC=8C=E5=88=86=E5=8C=BA=E5=8F=91=E9=80=81?=
 =?UTF-8?q?=E7=9A=84=E8=A1=A8pk=E4=B8=8D=E8=83=BD=E4=BF=AE=E6=94=B9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../otter/canal/protocol/FlatMessage.java     | 12 +---
 .../alibaba/otter/canal/protocol/Message.java | 58 +++++++------------
 2 files changed, 21 insertions(+), 49 deletions(-)

diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
index 53924d4a..f46c26a4 100644
--- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
+++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java
@@ -344,17 +344,7 @@ public class FlatMessage implements Serializable {
                     } else {
                         int idx = 0;
                         for (Map<String, String> row : flatMessage.getData()) {
-                            Map<String, String> o = null;
-                            if (flatMessage.getOld() != null) {
-                                o = flatMessage.getOld().get(idx);
-                            }
-                            String value;
-                            // 如果old中有pk值说明主键有修改, 以旧的主键值hash为准
-                            if (o != null && o.containsKey(pk)) {
-                                value = o.get(pk);
-                            } else {
-                                value = row.get(pk);
-                            }
+                            String value = row.get(pk);
                             if (value == null) {
                                 value = "";
                             }
diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java
index 6fc42a41..1013bfe6 100644
--- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java
+++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java
@@ -173,48 +173,30 @@ public class Message implements Serializable {
                         partitionEntries[0].add(entry);
                     }
                     for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
-                        boolean hasOldPk = false;
-                        // 如果before中有pk值说明主键有修改, 以旧的主键值hash为准
-                        for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
-                            CanalEntry.Column pkColumn = null;
-                            if (pk != null) {
-                                if (column.getName().equalsIgnoreCase(pk)) {
-                                    pkColumn = column;
-                                }
-                            } else {
-                                if (column.getIsKey()) {
-                                    pkColumn = column;
-                                }
-                            }
-                            if (pkColumn != null) {
-                                int hash = pkColumn.getValue().hashCode();
-                                int pkHash = Math.abs(hash) % partitionsNum;
-                                pkHash = Math.abs(pkHash);
-                                partitionEntries[pkHash].add(entry);
-                                hasOldPk = true;
-                                break;
-                            }
-                        }
-                        if (!hasOldPk) {
+                        String pkValue = null;
+                        if (pk != null) {
                             for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
-                                CanalEntry.Column pkColumn = null;
-                                if (pk != null) {
-                                    if (column.getName().equalsIgnoreCase(pk)) {
-                                        pkColumn = column;
-                                    }
-                                } else {
-                                    if (column.getIsKey()) {
-                                        pkColumn = column;
-                                    }
-                                }
-                                if (pkColumn != null) {
-                                    int hash = pkColumn.getValue().hashCode();
-                                    int pkHash = Math.abs(hash) % partitionsNum;
-                                    pkHash = Math.abs(pkHash);
-                                    partitionEntries[pkHash].add(entry);
+                                if (column.getName().equalsIgnoreCase(pk)) {
+                                    pkValue = column.getValue();
                                     break;
                                 }
                             }
+                        } else {
+                            for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
+                                if (column.getIsKey()) {
+                                    pkValue = column.getValue();
+                                    break;
+                                }
+                            }
+                        }
+                        if (pkValue != null) {
+                            int hash = pkValue.hashCode();
+                            int pkHash = Math.abs(hash) % partitionsNum;
+                            pkHash = Math.abs(pkHash);
+                            partitionEntries[pkHash].add(entry);
+                        } else {
+                            // 没有找到主键
+                            partitionEntries[0].add(entry);
                         }
                     }
                 }