From 96dca0b9e6b6dcc12fc439d2f17d85a4656485df Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 17 Nov 2016 17:52:52 +0800 Subject: [PATCH] fixed issue #136 , support rds no primary key && mysql.ha_health_check --- .../parse/driver/mysql/MysqlConnector.java | 2 +- .../otter/canal/parse/inbound/TableMeta.java | 12 +++ .../inbound/mysql/dbsync/LogEventConvert.java | 76 +++++++++++++------ .../inbound/mysql/dbsync/TableMetaCache.java | 17 +++++ 4 files changed, 84 insertions(+), 23 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java index f5d85915..6a5a4ae6 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java @@ -104,7 +104,7 @@ public class MysqlConnector { executor.update("KILL CONNECTION " + connectionId); } catch (Exception e) { // 忽略具体异常 - logger.warn("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e)); + logger.info("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e)); } finally { if (connector != null) { connector.disconnect(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java index 712919de..de934fbc 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.parse.inbound; +import java.util.ArrayList; import java.util.List; import org.apache.commons.lang.StringUtils; @@ -44,6 +45,17 @@ public class TableMeta { this.fileds = fileds; } + public List getPrimaryFields() { + List primarys = new ArrayList(); + for (FieldMeta meta : fileds) { + if (meta.isKey()) { + primarys.add(meta); + } + } + + return primarys; + } + public static class FieldMeta { private String columnName; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index e066dd96..1bbb0eb8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -85,7 +85,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92 private boolean filterTableError = false; // 新增rows过滤，用于仅订阅除rows以外的数据 - private boolean filterRows = false; + private boolean filterRows = false; public Entry parse(LogEvent logEvent) throws CanalParseException { if (logEvent == null || logEvent instanceof UnknownLogEvent) { @@ -326,6 +326,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return null; } + if (tableMetaCache.isOnRDS() && "mysql.ha_health_check".equals(fullname)) { + // 忽略rds模式的mysql.ha_health_check心跳数据 + return null; + } + EventType eventType = null; int type = event.getHeader().getType(); if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) { @@ -402,34 +407,51 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException { - final int columnCnt = event.getTable().getColumnCnt(); - final ColumnInfo[] columnInfo = event.getTable().getColumnInfo(); + int columnCnt = event.getTable().getColumnCnt(); + ColumnInfo[] columnInfo = event.getTable().getColumnInfo(); boolean tableError = false; // check table fileds count，只能处理加字段 + boolean existRDSNoPrimaryKey = false; if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) { - // online ddl增加字段操作步骤： - // 1. 新增一张临时表，将需要做ddl表的数据全量导入 - // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表 - // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作 - // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作 - // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作 - tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次 - if (tableMeta == null) { - tableError = true; - if (!filterTableError) { - throw new CanalParseException("not found [" + event.getTable().getDbName() + "." - + event.getTable().getTableName() + "] in db , pls check!"); + if (tableMetaCache.isOnRDS()) { + // 特殊处理下RDS的场景 + List primaryKeys = tableMeta.getPrimaryFields(); + if (primaryKeys == null || primaryKeys.isEmpty()) { + if (columnInfo.length == tableMeta.getFileds().size() + 1 + && columnInfo[columnInfo.length - 1].type == LogEvent.MYSQL_TYPE_LONGLONG) { + existRDSNoPrimaryKey = true; + } } } - // 在做一次判断 - if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) { - tableError = true; - if (!filterTableError) { - throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName() + "," - + columnInfo.length + " vs " + tableMeta.getFileds().size()); + if (!existRDSNoPrimaryKey) { + // online ddl增加字段操作步骤： + // 1. 新增一张临时表，将需要做ddl表的数据全量导入 + // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表 + // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作 + // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作 + // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作 + tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false);// 强制重新获取一次 + if (tableMeta == null) { + tableError = true; + if (!filterTableError) { + throw new CanalParseException("not found [" + event.getTable().getDbName() + "." + + event.getTable().getTableName() + "] in db , pls check!"); + } } + + // 在做一次判断 + if (tableMeta != null && columnInfo.length > tableMeta.getFileds().size()) { + tableError = true; + if (!filterTableError) { + throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName() + + "," + columnInfo.length + " vs " + tableMeta.getFileds().size()); + } + } + } else { + logger.warn("[" + event.getTable().getDbName() + "." + event.getTable().getTableName() + + "] is no primary key , skip alibaba_rds_row_id column"); } } @@ -440,6 +462,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar continue; } + if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { + // 不解析最后一列 + buffer.nextValue(info.type, info.meta, false); + continue; + } + Column.Builder columnBuilder = Column.newBuilder(); FieldMeta fieldMeta = null; @@ -465,6 +493,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } buffer.nextValue(info.type, info.meta, isBinary); + if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { + // 不解析最后一列 + continue; + } int javaType = buffer.getJavaType(); if (buffer.isNull()) { @@ -718,7 +750,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterTableError(boolean filterTableError) { this.filterTableError = filterTableError; } - + public void setFilterRows(boolean filterRows) { this.filterRows = filterRows; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java index 0bf22791..d8575ec8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java @@ -32,6 +32,7 @@ public class TableMetaCache { public static final String COLUMN_DEFAULT = "COLUMN_DEFAULT"; public static final String EXTRA = "EXTRA"; private MysqlConnection connection; + private boolean isOnRDS = false; // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表 private Map tableMetaCache; @@ -56,6 +57,13 @@ public class TableMetaCache { }); + try { + ResultSetPacket packet = connection.query("show global variables like 'rds\\_%'"); + if (packet.getFieldValues().size() > 0) { + isOnRDS = true; + } + } catch (IOException e) { + } } public TableMeta getTableMeta(String schema, String table) { @@ -136,4 +144,13 @@ public class TableMetaCache { .append('`') .toString(); } + + public boolean isOnRDS() { + return isOnRDS; + } + + public void setOnRDS(boolean isOnRDS) { + this.isOnRDS = isOnRDS; + } + }