chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed delete message send mq bugfix

Browse Source
This commit is contained in:
agapple
2019-05-28 00:14:20 +08:00
parent e808c6ec70
commit 058e1fc546
1 changed files with 12 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
View File

@@ -222,35 +222,28 @@ public class MQMessageUtils {
} else { } else {
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
int hashCode = database.hashCode(); int hashCode = database.hashCode();
CanalEntry.EventType eventType = rowChange.getEventType();
List<CanalEntry.Column> columns = null;
if (eventType == CanalEntry.EventType.DELETE) {
columns = rowData.getBeforeColumnsList();
} else {
columns = rowData.getAfterColumnsList();
}
if (hashMode.autoPkHash) { if (hashMode.autoPkHash) {
// isEmpty use default pkNames // isEmpty use default pkNames
for (CanalEntry.Column column : rowData.getAfterColumnsList()) { for (CanalEntry.Column column : columns) {
if (column.getIsKey()) { if (column.getIsKey()) {
hashCode = hashCode ^ column.getValue().hashCode(); hashCode = hashCode ^ column.getValue().hashCode();
} }
} }
} else { } else {
try { for (CanalEntry.Column column : columns) {
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())) { if (checkPkNamesHasContain(hashMode.pkNames, column.getName())) {
hashCode = hashCode ^ column.getValue().hashCode(); 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();
}
}
int pkHash = Math.abs(hashCode) % partitionsNum; int pkHash = Math.abs(hashCode) % partitionsNum;
pkHash = Math.abs(pkHash); pkHash = Math.abs(pkHash);