删除操作计算hash值的情况没有考虑，导致删除时的分区和插入时的分区不一致 (#1848)

alibaba · May 27, 2019 · 37fc248 · 37fc248
1 parent d0a83aa
commit 37fc248
Showing 1 changed file with 18 additions and 3 deletions.
diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
@@ -230,10 +230,25 @@ public static Message[] messagePartition(Message message, Integer partitionsNum,
                                     }
                                 }
                             } else {
-                                for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
-                                    if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) {
-                                        hashCode = hashCode ^ column.getValue().hashCode();
+                                try {
+                                    CanalEntry.EventType eventType = RowChange.parseFrom(entry.getStoreValue()).getEventType();
+                                    if(eventType == CanalEntry.EventType.DELETE){
+                                        for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
+                                            if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) {
+                                                hashCode = hashCode ^ column.getValue().hashCode();
+                                            }
+                                        }
                                     }
+                                    else {
+                                        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
+                                            if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) {
+                                                hashCode = hashCode ^ column.getValue().hashCode();
+                                            }
+                                        }
+                                    }
+                                }
+                                catch (InvalidProtocolBufferException e) {
+                                    e.printStackTrace();
                                 }
                             }