diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index cbcc19f8..df6f49e2 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -40,6 +40,7 @@ canal.instance.filter.query.dcl = false canal.instance.filter.query.dml = false canal.instance.filter.query.ddl = false canal.instance.filter.table.error = false +canal.instance.filter.rows = false # binlog format/image check canal.instance.binlog.format = ROW,STATEMENT,MIXED @@ -63,4 +64,4 @@ canal.instance.global.lazy = false #canal.instance.global.manager.address = 127.0.0.1:1099 #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml -#canal.instance.global.spring.xml = classpath:spring/default-instance.xml \ No newline at end of file +#canal.instance.global.spring.xml = classpath:spring/default-instance.xml diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index f12b72d3..4e092257 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -178,8 +178,9 @@ + - \ No newline at end of file + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index 618beed1..191d0c11 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -163,8 +163,9 @@ + - \ No newline at end of file + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index c135196b..6fefaa87 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -250,8 +250,9 @@ + - \ No newline at end of file + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index b7afc130..269b9deb 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -151,8 +151,9 @@ + - \ No newline at end of file + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 85644dcd..ebc7bcd4 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -22,6 +22,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected boolean filterQueryDcl = false; protected boolean filterQueryDml = false; protected boolean filterQueryDdl = false; + protected boolean filterRows = false; protected boolean filterTableError = false; protected BinlogParser buildParser() { @@ -38,6 +39,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { convert.setFilterQueryDcl(filterQueryDcl); convert.setFilterQueryDml(filterQueryDml); convert.setFilterQueryDdl(filterQueryDdl); + convert.setFilterRows(filterRows); convert.setFilterTableError(filterTableError); return convert; } @@ -86,6 +88,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { public void setFilterQueryDdl(boolean filterQueryDdl) { this.filterQueryDdl = filterQueryDdl; } + + public void setFilterRows(boolean filterRows) { + this.filterRows = filterRows; + } public void setFilterTableError(boolean filterTableError) { this.filterTableError = filterTableError; 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 cc092d46..37ea0d00 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 @@ -84,6 +84,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private boolean filterQueryDdl = false; // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92 private boolean filterTableError = false; + // 新增rows过滤,用于仅订阅除rows以外的数据 + private boolean filterRows = false; public Entry parse(LogEvent logEvent) throws CanalParseException { if (logEvent == null || logEvent instanceof UnknownLogEvent) { @@ -305,6 +307,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } private Entry parseRowsEvent(RowsLogEvent event) { + if (filterRows) { + return null; + } try { TableMapLogEvent table = event.getTable(); if (table == null) { @@ -713,5 +718,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterTableError(boolean filterTableError) { this.filterTableError = filterTableError; } + + public void setFilterRows(boolean filterRows) { + this.filterRows = filterRows; + } }