From f9f810226dae3dcd47032e2f63d30e169395a10d Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Mon, 11 Sep 2023 10:57:43 +0800 Subject: [PATCH] fixed issue #4869 , support admin tsdb config --- .../resources/spring/default-instance.xml | 3 + .../main/resources/spring/file-instance.xml | 3 + .../main/resources/spring/memory-instance.xml | 3 + .../manager/CanalInstanceWithManager.java | 26 ++++----- .../manager/PlainCanalInstanceGenerator.java | 1 + .../mysql/AbstractMysqlEventParser.java | 56 +++++++++++++------ 6 files changed, 61 insertions(+), 31 deletions(-) diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 5a8f7813..6fe11744 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -185,6 +185,9 @@ + + + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index ebe97c3f..738d19c9 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -171,6 +171,9 @@ + + + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index ab667347..5c5c6d37 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -159,6 +159,9 @@ + + + diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 4cccb62b..55c0141f 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -24,13 +24,7 @@ import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.instance.core.AbstractCanalInstance; import com.alibaba.otter.canal.instance.manager.model.Canal; import com.alibaba.otter.canal.instance.manager.model.CanalParameter; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.HAMode; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode; -import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode; +import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*; import com.alibaba.otter.canal.meta.FileMixedMetaManager; import com.alibaba.otter.canal.meta.MemoryMetaManager; import com.alibaba.otter.canal.meta.PeriodMixedMetaManager; @@ -46,12 +40,7 @@ import com.alibaba.otter.canal.parse.inbound.mysql.rds.RdsBinlogEventParserProxy import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFactory; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder; -import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; -import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager; -import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager; -import com.alibaba.otter.canal.parse.index.MetaLogPositionManager; -import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager; -import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager; +import com.alibaba.otter.canal.parse.index.*; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.alibaba.otter.canal.sink.entry.EntryEventSink; @@ -338,6 +327,8 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { mysqlEventParser.setTsdbSnapshotExpire(parameters.getTsdbSnapshotExpire()); } boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable()); + // manager启动模式默认使用mysql tsdb机制 + final String tsdbSpringXml = "classpath:spring/tsdb/mysql-tsdb.xml"; if (tsdbEnable) { mysqlEventParser.setTableMetaTSDBFactory(new DefaultTableMetaTSDBFactory() { @@ -353,9 +344,12 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { System.setProperty("canal.instance.tsdb.url", parameters.getTsdbJdbcUrl()); System.setProperty("canal.instance.tsdb.dbUsername", parameters.getTsdbJdbcUserName()); System.setProperty("canal.instance.tsdb.dbPassword", parameters.getTsdbJdbcPassword()); + System.setProperty("canal.instance.destination", destination); - return TableMetaTSDBBuilder.build(destination, "classpath:spring/tsdb/mysql-tsdb.xml"); + return TableMetaTSDBBuilder.build(destination, tsdbSpringXml); } finally { + // reset + System.setProperty("canal.instance.destination", ""); System.setProperty("canal.instance.tsdb.url", ""); System.setProperty("canal.instance.tsdb.dbUsername", ""); System.setProperty("canal.instance.tsdb.dbPassword", ""); @@ -363,6 +357,10 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { } } }); + mysqlEventParser.setTsdbJdbcUrl(parameters.getTsdbJdbcUrl()); + mysqlEventParser.setTsdbJdbcUserName(parameters.getTsdbJdbcUserName()); + mysqlEventParser.setTsdbJdbcPassword(parameters.getTsdbJdbcPassword()); + mysqlEventParser.setTsdbSpringXml(tsdbSpringXml); mysqlEventParser.setEnableTsdb(tsdbEnable); } eventParser = mysqlEventParser; diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java index 61eb964b..b909f2b0 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/PlainCanalInstanceGenerator.java @@ -61,6 +61,7 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator { throw new CanalException(e); } finally { System.setProperty("canal.instance.destination", ""); + com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer.propertiesLocal.remove(); } } } 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 a8768131..802ce8dc 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 @@ -16,7 +16,6 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DefaultTableMetaTSDBFact import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory; import com.alibaba.otter.canal.protocol.position.EntryPosition; -import org.apache.commons.lang.StringUtils; public abstract class AbstractMysqlEventParser extends AbstractEventParser { @@ -24,6 +23,9 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected TableMetaTSDBFactory tableMetaTSDBFactory = new DefaultTableMetaTSDBFactory(); protected boolean enableTsdb = false; + protected String tsdbJdbcUrl; + protected String tsdbJdbcUserName; + protected String tsdbJdbcPassword; protected int tsdbSnapshotInterval = 24; protected int tsdbSnapshotExpire = 360; protected String tsdbSpringXml; @@ -149,14 +151,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { if (enableTsdb) { if (tableMetaTSDB == null) { synchronized (CanalEventParser.class) { - try { - // 设置当前正在加载的通道，加载spring查找文件时会用到该变量 - System.setProperty("canal.instance.destination", destination); - // 初始化 - tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml); - } finally { - System.setProperty("canal.instance.destination", ""); - } + buildTableMetaTSDB(tsdbSpringXml); } } } @@ -173,6 +168,28 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { super.stop(); } + protected void buildTableMetaTSDB(String tsdbSpringXml) { + if (tableMetaTSDB != null) { + return; + } + + try { + // 设置当前正在加载的通道，加载spring查找文件时会用到该变量 + System.setProperty("canal.instance.tsdb.url", tsdbJdbcUrl); + System.setProperty("canal.instance.tsdb.dbUsername", tsdbJdbcUserName); + System.setProperty("canal.instance.tsdb.dbPassword", tsdbJdbcPassword); + System.setProperty("canal.instance.destination", destination); + // 初始化 + this.tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml); + } finally { + // reset + System.setProperty("canal.instance.destination", ""); + System.setProperty("canal.instance.tsdb.url", ""); + System.setProperty("canal.instance.tsdb.dbUsername", ""); + System.setProperty("canal.instance.tsdb.dbPassword", ""); + } + } + protected MultiStageCoprocessor buildMultiStageCoprocessor() { MysqlMultiStageCoprocessor mysqlMultiStageCoprocessor = new MysqlMultiStageCoprocessor(parallelBufferSize, parallelThreadSize, @@ -256,20 +273,14 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { public void setEnableTsdb(boolean enableTsdb) { this.enableTsdb = enableTsdb; if (this.enableTsdb) { - if (tableMetaTSDB == null) { - // 初始化 - tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml); - } + buildTableMetaTSDB(tsdbSpringXml); } } public void setTsdbSpringXml(String tsdbSpringXml) { this.tsdbSpringXml = tsdbSpringXml; if (this.enableTsdb) { - if (tableMetaTSDB == null) { - // 初始化 - tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml); - } + buildTableMetaTSDB(tsdbSpringXml); } } @@ -301,4 +312,15 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { this.tsdbSnapshotExpire = tsdbSnapshotExpire; } + public void setTsdbJdbcUrl(String tsdbJdbcUrl) { + this.tsdbJdbcUrl = tsdbJdbcUrl; + } + + public void setTsdbJdbcUserName(String tsdbJdbcUserName) { + this.tsdbJdbcUserName = tsdbJdbcUserName; + } + + public void setTsdbJdbcPassword(String tsdbJdbcPassword) { + this.tsdbJdbcPassword = tsdbJdbcPassword; + } }