diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index d1e81ec0..0b112757 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -52,6 +52,8 @@ public class CanalStater { // disable netty System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true"); String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN); + // 设置为raw避免ByteString->Entry的二次解析 + System.setProperty("canal.instance.memory.rawEntry", "false"); if ("true".equals(autoScan)) { String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR); if (StringUtils.isEmpty(rootDir)) { @@ -68,13 +70,13 @@ public class CanalStater { }); if (instanceDirs != null && instanceDirs.length > 0) { List instances = Lists.transform(Arrays.asList(instanceDirs), - new Function() { + new Function() { - @Override - public String apply(File instanceDir) { - return instanceDir.getName(); - } - }); + @Override + public String apply(File instanceDir) { + return instanceDir.getName(); + } + }); System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances)); } } diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 65a8c6d2..bc1db836 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -46,7 +46,7 @@ canal.instance.filter.black.regex= # mq config canal.mq.topic=example # dynamic topic route by schema or table regex -#canal.mq.dynamicTopic=.*,mytest,mytest\\..*,mytest2.user +#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..* canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=4 diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 338bd962..ee3408e9 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -10,7 +10,6 @@ import java.util.List; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; -import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.math.NumberUtils; import org.slf4j.Logger; @@ -21,6 +20,7 @@ import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor; import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket; @@ -131,9 +131,11 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); - // 若entry position存在gtid，则使用传入的gtid作为gtidSet 拼接的标准,否则同时开启gtid和tsdb时，会导致丢失gtid + // 若entry position存在gtid，则使用传入的gtid作为gtidSet + // 拼接的标准,否则同时开启gtid和tsdb时，会导致丢失gtid // 而当源端数据库gtid 有purged时会有如下类似报错 - // 'errno = 1236, sqlstate = HY000 errmsg = The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1 ... + // 'errno = 1236, sqlstate = HY000 errmsg = The slave is connecting + // using CHANGE MASTER TO MASTER_AUTO_POSITION = 1 ... if (StringUtils.isNotEmpty(gtid)) { decoder.handle(LogEvent.GTID_LOG_EVENT); context.setGtidSet(MysqlGTIDSet.parse(gtid)); @@ -417,7 +419,9 @@ public class MysqlConnection implements ErosaConnection { // '@@global.binlog_checksum'需要去掉单引号,在mysql 5.6.29下导致master退出 update("set @master_binlog_checksum= @@global.binlog_checksum"); } catch (Exception e) { - logger.warn("update master_binlog_checksum failed", e); + if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) { + logger.warn("update master_binlog_checksum failed", e); + } } try { @@ -434,7 +438,9 @@ public class MysqlConnection implements ErosaConnection { // mariadb针对特殊的类型，需要设置session变量 update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'"); } catch (Exception e) { - logger.warn("update mariadb_slave_capability failed", e); + if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) { + logger.warn("update mariadb_slave_capability failed", e); + } } /** diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index f4a23dc8..3ae33b9d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -77,9 +77,7 @@ public class MemoryTableMeta implements TableMetaTSDB { if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush") && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant") && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create user") - && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop user") - && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create database") - && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop database")) { + && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop user")) { repository.console(ddl); } } catch (Throwable e) { diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java index bf0deafe..52aa7f34 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java @@ -18,19 +18,12 @@ public class FastsqlSchemaTest { @Test public void testSimple() throws FileNotFoundException, IOException { SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL); - String sql = "create table yushitai_test.card_record ( id bigint auto_increment, name varchar(32) DEFAULT NULL) auto_increment=256 " - + "ALTER TABLE yushitai_test.card_record ADD COLUMN remark2 varchar(255) DEFAULT NULL , ALGORITHM=inplace,LOCK=NONE;" - + "ALTER TABLE yushitai_test.card_record modify COLUMN name varchar(64) DEFAULT NULL , ALGORITHM=copy,LOCK=SHARED; " - + "alter table yushitai_test.card_record add index index_name(name) ;" - + "alter table yushitai_test.card_record add index index_name(name) ;" - + "alter table yushitai_test.card_record add Constraint pk_id PRIMARY KEY (id);" - + "alter table yushitai_test.card_record add Constraint pk_id PRIMARY KEY (id);" - + "alter table yushitai_test.card_record add Constraint UNIQUE index uk_name(name);" - + "alter table yushitai_test.card_record add Constraint UNIQUE index uk_name(name);"; + String sql = "create table quniya4(name varchar(255) null,value varchar(255) null,id int not null,constraint quniya4_pk primary key (id));" + + "alter table quniya4 modify id int not null first;"; repository.console(sql); - repository.setDefaultSchema("yushitai_test"); - SchemaObject table = repository.findTable("card_record"); + repository.setDefaultSchema("test"); + SchemaObject table = repository.findTable("quniya4"); System.out.println(table.getStatement().toString()); } } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java index fa0999fe..2afd57e5 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java @@ -18,7 +18,7 @@ import com.alibaba.otter.canal.parse.inbound.TableMeta; * @author agapple 2017年8月1日 下午7:15:54 */ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) +@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" }) public class MemoryTableMeta_DDL_Test { @Test @@ -52,4 +52,18 @@ public class MemoryTableMeta_DDL_Test { Assert.assertEquals(meta.getFieldMetaByName("id").isKey(), true); Assert.assertEquals(meta.getFieldMetaByName("name").isUnique(), true); } + + @Test + public void test3() throws Throwable { + MemoryTableMeta memoryTableMeta = new MemoryTableMeta(); + URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); + File dummyFile = new File(url.getFile()); + File create = new File(dummyFile.getParent() + "/ddl", "ddl_test3.sql"); + String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n"); + memoryTableMeta.apply(null, "test", sql, null); + + TableMeta meta = memoryTableMeta.find("test", "quniya4"); + System.out.println(meta); + Assert.assertTrue(meta.getFields().get(0).getColumnName().equalsIgnoreCase("id")); + } } diff --git a/parse/src/test/resources/ddl/ddl_test3.sql b/parse/src/test/resources/ddl/ddl_test3.sql new file mode 100644 index 00000000..d2544f32 --- /dev/null +++ b/parse/src/test/resources/ddl/ddl_test3.sql @@ -0,0 +1,3 @@ +create table quniya4(name varchar(255) null,value varchar(255) null,id int not null,constraint quniya4_pk primary key (id)); +alter table quniya4 modify id int not null first; +alter table quniya4 modify id int auto_increment; \ No newline at end of file diff --git a/pom.xml b/pom.xml index 5de04ea4..61d91a28 100644 --- a/pom.xml +++ b/pom.xml @@ -246,7 +246,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_644 + 2.0.0_preview_836 com.alibaba 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 index 370b848d..17568d44 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -509,7 +509,7 @@ public class MQMessageUtils { } private static Set matchTopics(String name, String dynamicTopicConfigs) { - String[] router = StringUtils.split(dynamicTopicConfigs, ';'); + String[] router = StringUtils.split(dynamicTopicConfigs, ','); Set topics = new HashSet<>(); for (String item : router) { int i = item.indexOf(":"); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index ae929b41..796a60b5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -53,9 +53,6 @@ public class CanalMQStarter { System.setProperty("canal.instance.filter.transaction.entry", "true"); } - // 设置为raw避免ByteString->Entry的二次解析 - System.setProperty("canal.instance.memory.rawEntry", "false"); - canalServer = CanalServerWithEmbedded.instance(); // 对应每个instance启动一个worker线程