fixed h2 support

deployer/src/main/resources/canal.properties

@@ -60,7 +60,8 @@ canal.conf.dir = ../conf
 canal.auto.scan = true
 canal.auto.scan.interval = 5
 
-canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml
+canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml
+#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml
 
 canal.instance.global.mode = spring 
 canal.instance.global.lazy = false

deployer/src/main/resources/example/instance.properties

@@ -4,13 +4,15 @@ canal.instance.mysql.slaveId=1234
 # position info
 canal.instance.master.address=127.0.0.1:3306
 canal.instance.master.journal.name=mysql-bin.000001
-canal.instance.master.position=68240
+canal.instance.master.position=104606
 canal.instance.master.timestamp=
 
 
 # table meta tsdb info
 canal.instance.tsdb.enable=true
-canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
+canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
+canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
+#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
 canal.instance.tsdb.dbUsername=canal
 canal.instance.tsdb.dbPassword=canal
 

deployer/src/main/resources/spring/tsdb/h2-tsdb.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
+	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:lang="http://www.springframework.org/schema/lang"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+           http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
+           http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.0.xsd
+           http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
+           http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"
+	default-autowire="byName">
+	
+	<!-- properties -->
+	<bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
+		<property name="ignoreResourceNotFound" value="true" />
+		<property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
+		<property name="locationNames">
+			<list>
+				<value>classpath:canal.properties</value>
+				<value>classpath:${canal.instance.destination:}/instance.properties</value>
+			</list>
+		</property>
+	</bean>
+	
+	<!-- 基于db的实现 -->
+	<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager">
+		<property name="metaHistoryDAO" ref="metaHistoryDAO"/>
+		<property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
+	</bean>
+	
+    <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
+        <property name="driverClassName" value="org.h2.Driver" />
+		<property name="url" value="${canal.instance.tsdb.url:}" />
+		<property name="username" value="${canal.instance.tsdb.dbUsername:canal}" />
+		<property name="password" value="${canal.instance.tsdb.dbPassword:canal}" />
+      	<property name="maxActive" value="30" />
+        <property name="initialSize" value="0" />
+        <property name="minIdle" value="1" />
+        <property name="maxWait" value="10000" />
+        <property name="timeBetweenEvictionRunsMillis" value="60000" />
+        <property name="minEvictableIdleTimeMillis" value="300000" />
+        <property name="testWhileIdle" value="true" />
+        <property name="testOnBorrow" value="false" />
+        <property name="testOnReturn" value="false" />
+        <property name="useUnfairLock" value="true" />
+	</bean>
+
+    <bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean">
+        <property name="dataSource" ref="dataSource"/>
+        <property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/>
+    </bean>
+
+    <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
+        <property name="sqlMapClient" ref="sqlMapClient"/>
+    </bean>
+
+    <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
+        <property name="sqlMapClient" ref="sqlMapClient"/>
+    </bean>
+</beans>

parse/pom.xml

@@ -58,6 +58,10 @@
 			<groupId>org.apache.ibatis</groupId>
 			<artifactId>ibatis-sqlmap</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.h2database</groupId>
+			<artifactId>h2</artifactId>
+		</dependency>
 		<!-- test dependency -->
 		<dependency>
 			<groupId>junit</groupId>

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManager.java

@@ -46,10 +46,11 @@ public class TableMetaManager implements TableMetaTSDB {
 
     private static Logger              logger        = LoggerFactory.getLogger(TableMetaManager.class);
     private static Pattern             pattern       = Pattern.compile("Duplicate entry '.*' for key '*'");
+    private static Pattern             h2Pattern     = Pattern.compile("Unique index or primary key violation");
     private static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L);
     private String                     destination;
     private MemoryTableMeta            memoryTableMeta;
-    private MysqlConnection            connection;                                                         // 查询meta信息的链接
+    private MysqlConnection            connection;                                                              // 查询meta信息的链接
     private CanalEventFilter           filter;
     private CanalEventFilter           blackFilter;
     private EntryPosition              lastPosition;
@@ -466,7 +467,7 @@ public class TableMetaManager implements TableMetaTSDB {
     }
 
     public boolean isUkDuplicateException(Throwable t) {
-        if (pattern.matcher(t.getMessage()).find()) {
+        if (pattern.matcher(t.getMessage()).find() || h2Pattern.matcher(t.getMessage()).find()) {
             // 违反外键约束时也抛出这种异常，所以这里还要判断包含字符串Duplicate entry
             return true;
         }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaBaseDAO.java

@@ -0,0 +1,59 @@
+package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao;
+
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import javax.sql.DataSource;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport;
+
+/**
+ * @author agapple 2017年10月14日 上午1:05:22
+ * @since 1.0.25
+ */
+@SuppressWarnings("deprecation")
+public class MetaBaseDAO extends SqlMapClientDaoSupport {
+
+    protected boolean isH2 = false;
+
+    protected void initTable(String tableName) throws Exception {
+        Connection conn = null;
+        InputStream input = null;
+        try {
+            DataSource dataSource = getDataSource();
+            conn = dataSource.getConnection();
+            String name = "mysql";
+            isH2 = isH2(conn);
+            if (isH2) {
+                name = "h2";
+            }
+            input = Thread.currentThread()
+                .getContextClassLoader()
+                .getResourceAsStream("ddl/" + name + "/" + tableName + ".sql");
+            if (input == null) {
+                return;
+            }
+
+            String sql = StringUtils.join(IOUtils.readLines(input), "\n");
+            Statement stmt = conn.createStatement();
+            stmt.execute(sql);
+            stmt.close();
+        } catch (Throwable e) {
+            logger.warn("init " + tableName + " failed", e);
+        } finally {
+            IOUtils.closeQuietly(input);
+            if (conn != null) {
+                conn.close();
+            }
+        }
+    }
+
+    private boolean isH2(Connection conn) throws SQLException {
+        String product = conn.getMetaData().getDatabaseProductName();
+        return StringUtils.containsIgnoreCase(product, "H2");
+    }
+}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java

@@ -1,19 +1,10 @@
 package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao;
 
-import java.io.InputStream;
-import java.sql.Connection;
-import java.sql.Statement;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 
-import javax.sql.DataSource;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.StringUtils;
-import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport;
-
 import com.google.common.collect.Maps;
 
 /**
@@ -23,7 +14,7 @@ import com.google.common.collect.Maps;
  * @since 3.2.5
  */
 @SuppressWarnings("deprecation")
-public class MetaHistoryDAO extends SqlMapClientDaoSupport {
+public class MetaHistoryDAO extends MetaBaseDAO {
 
     public Long insert(MetaHistoryDO metaDO) {
         return (Long) getSqlMapClientTemplate().insert("meta_history.insert", metaDO);
@@ -56,27 +47,6 @@ public class MetaHistoryDAO extends SqlMapClientDaoSupport {
     }
 
     protected void initDao() throws Exception {
-        Connection conn = null;
-        InputStream input = null;
-        try {
-            DataSource dataSource = getDataSource();
-            conn = dataSource.getConnection();
-            input = Thread.currentThread().getContextClassLoader().getResourceAsStream("ddl/mysql/meta_history.sql");
-            if (input == null) {
-                return;
-            }
-
-            String sql = StringUtils.join(IOUtils.readLines(input), "\n");
-            Statement stmt = conn.createStatement();
-            stmt.execute(sql);
-            stmt.close();
-        } catch (Throwable e) {
-            logger.warn("init meta_history failed", e);
-        } finally {
-            IOUtils.closeQuietly(input);
-            if (conn != null) {
-                conn.close();
-            }
-        }
+        initTable("meta_history");
     }
 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java

@@ -1,18 +1,9 @@
 package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao;
 
-import java.io.InputStream;
-import java.sql.Connection;
-import java.sql.Statement;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
 
-import javax.sql.DataSource;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.StringUtils;
-import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport;
-
 import com.google.common.collect.Maps;
 
 /**
@@ -22,7 +13,7 @@ import com.google.common.collect.Maps;
  * @since 3.2.5
  */
 @SuppressWarnings("deprecation")
-public class MetaSnapshotDAO extends SqlMapClientDaoSupport {
+public class MetaSnapshotDAO extends MetaBaseDAO {
 
     public Long insert(MetaSnapshotDO snapshotDO) {
         return (Long) getSqlMapClientTemplate().insert("meta_snapshot.insert", snapshotDO);
@@ -59,27 +50,7 @@ public class MetaSnapshotDAO extends SqlMapClientDaoSupport {
     }
 
     protected void initDao() throws Exception {
-        Connection conn = null;
-        InputStream input = null;
-        try {
-            DataSource dataSource = getDataSource();
-            conn = dataSource.getConnection();
-            input = Thread.currentThread().getContextClassLoader().getResourceAsStream("ddl/mysql/meta_snapshot.sql");
-            if (input == null) {
-                return;
-            }
-
-            String sql = StringUtils.join(IOUtils.readLines(input), "\n");
-            Statement stmt = conn.createStatement();
-            stmt.execute(sql);
-            stmt.close();
-        } catch (Throwable e) {
-            logger.warn("init meta_history failed", e);
-        } finally {
-            IOUtils.closeQuietly(input);
-            if (conn != null) {
-                conn.close();
-            }
-        }
+        initTable("meta_snapshot");
     }
+
 }

parse/src/main/resources/ddl/h2/meta_history.sql

@@ -0,0 +1,21 @@
+CREATE TABLE IF NOT EXISTS `meta_history` (
+  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `gmt_create` datetime NOT NULL COMMENT '创建时间',
+  `gmt_modified` datetime NOT NULL COMMENT '修改时间',
+  `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
+  `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
+  `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
+  `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
+  `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
+  `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema',
+  `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema',
+  `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table',
+  `sql_text` longtext DEFAULT NULL COMMENT '执行的sql',
+  `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型',
+  `extra` text DEFAULT NULL COMMENT '额外的扩展信息',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY meta_history_binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
+  KEY `meta_history_destination` (`destination`),
+  KEY `meta_history_destination_timestamp` (`destination`,`binlog_timestamp`),
+  KEY `meta_history_gmt_modified` (`gmt_modified`)
+);

parse/src/main/resources/ddl/h2/meta_snapshot.sql

@@ -0,0 +1,17 @@
+CREATE TABLE IF NOT EXISTS `meta_snapshot` (
+  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
+  `gmt_create` datetime NOT NULL COMMENT '创建时间',
+  `gmt_modified` datetime NOT NULL COMMENT '修改时间',
+  `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
+  `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
+  `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
+  `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
+  `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
+  `data` longtext DEFAULT NULL COMMENT '表结构数据',
+  `extra` text DEFAULT NULL COMMENT '额外的扩展信息',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY meta_snapshot_binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
+  KEY `meta_snapshot_destination` (`destination`),
+  KEY `meta_snapshot_destination_timestamp` (`destination`,`binlog_timestamp`),
+  KEY `meta_snapshot_gmt_modified` (`gmt_modified`)
+);

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java

@@ -36,7 +36,7 @@ public class MysqlDumpTest {
         controller.setMasterPosition(startPosition);
         controller.setEnableTsdb(true);
         controller.setDestination("example");
-        controller.setTsdbSpringXml("classpath:tsdb/mysql-tsdb.xml");
+        controller.setTsdbSpringXml("classpath:tsdb/h2-tsdb.xml");
         controller.setEventFilter(new AviaterRegexFilter("test\\..*"));
         controller.setEventBlackFilter(new AviaterRegexFilter("canal_tsdb\\..*"));
         controller.setEventSink(new AbstractCanalEventSinkTest<List<Entry>>() {

parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerTest.java

@@ -22,7 +22,7 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition;
  * @since 3.2.5
  */
 @RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" })
+@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" })
 public class TableMetaManagerTest {
 
     @Resource
@@ -40,8 +40,7 @@ public class TableMetaManagerTest {
         tableMetaManager.apply(position, "tddl5_00", createSql, null);
 
         String alterSql = "alter table `test` add column name varchar(32) after c_varchar";
-        position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927853L, 1501660815000L);
+        position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927854L, 1501660815000L);
         tableMetaManager.apply(position, "tddl5_00", alterSql, null);
-
     }
 }

parse/src/test/resources/tsdb/h2-tsdb.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns="http://www.springframework.org/schema/beans" xmlns:tx="http://www.springframework.org/schema/tx"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+	http://www.springframework.org/schema/tx
+    http://www.springframework.org/schema/tx/spring-tx-2.0.xsd"
+       default-autowire="byName">
+	
+	<!-- 基于db的实现 -->
+	<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager">
+		<property name="metaHistoryDAO" ref="metaHistoryDAO"/>
+		<property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
+	</bean>
+	
+    <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
+        <property name="driverClassName" value="org.h2.Driver" />
+        <property name="url" value="jdbc:h2:/tmp/example;CACHE_SIZE=1000;MODE=MYSQL;" />
+		<property name="username" value="canal" />
+		<property name="password" value="canal" />
+        <property name="maxActive" value="30" />
+        <property name="initialSize" value="0" />
+        <property name="minIdle" value="1" />
+        <property name="maxWait" value="10000" />
+        <property name="timeBetweenEvictionRunsMillis" value="60000" />
+        <property name="minEvictableIdleTimeMillis" value="300000" />
+        <property name="testWhileIdle" value="true" />
+        <property name="testOnBorrow" value="false" />
+        <property name="testOnReturn" value="false" />
+        <property name="useUnfairLock" value="true" />
+	</bean>
+
+    <bean id="sqlMapClient" class="org.springframework.orm.ibatis.SqlMapClientFactoryBean">
+        <property name="dataSource" ref="dataSource"/>
+        <property name="configLocation" value="classpath:tsdb/sql-map/sqlmap-config.xml"/>
+    </bean>
+
+    <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
+        <property name="sqlMapClient" ref="sqlMapClient"/>
+    </bean>
+
+    <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
+        <property name="sqlMapClient" ref="sqlMapClient"/>
+    </bean>
+</beans>

pom.xml

@@ -262,6 +262,11 @@
                 <artifactId>slf4j-api</artifactId>
                 <version>1.7.12</version>
             </dependency>
+            <dependency>
+                <groupId>com.h2database</groupId>
+                <artifactId>h2</artifactId>
+                <version>1.4.196</version>
+            </dependency>
             <!-- test dependency -->
             <dependency>
                 <groupId>junit</groupId>