diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java index d0c6746d..27ecf3db 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java @@ -68,7 +68,7 @@ public class BaseLogFetcherTest { // update需要处理before/after System.out.println("-------> before"); parseOneRow(event, buffer, columns, false); - if (!buffer.nextOneRow(changeColumns)) { + if (!buffer.nextOneRow(changeColumns, true)) { break; } System.out.println("-------> after"); @@ -97,7 +97,7 @@ public class BaseLogFetcherTest { } ColumnInfo info = columnInfo[i]; - buffer.nextValue(info.type, info.meta); + buffer.nextValue(null , i ,info.type, info.meta); if (buffer.isNull()) { // diff --git a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml index beb61236..c05b5200 100644 --- a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml +++ b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml @@ -23,7 +23,7 @@ - + diff --git a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml index 58cab638..87f407b3 100644 --- a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml +++ b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml @@ -23,7 +23,7 @@ - + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java index d82e5eff..68f1ff5a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java @@ -71,6 +71,7 @@ public class DruidDdlParser { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), true); processName(ddlResult, schmeaName, ((SQLAlterTableRename) item).getToName(), false); + ddlResult.setType(EventType.RENAME); ddlResults.add(ddlResult); } else if (item instanceof SQLAlterTableAddIndex) { DdlResult ddlResult = new DdlResult(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index b0dd89ce..8896e2f5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -7,6 +7,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; @@ -48,18 +49,26 @@ public class DatabaseTableMeta implements TableMetaTSDB { private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.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 ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]"); + thread.setDaemon(true); + return thread; + } + }); private String destination; private MemoryTableMeta memoryTableMeta; private MysqlConnection connection; // 查询meta信息的链接 private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; - private ScheduledExecutorService scheduler; private MetaHistoryDAO metaHistoryDAO; private MetaSnapshotDAO metaSnapshotDAO; private int snapshotInterval = 24; private int snapshotExpire = 360; - + private ScheduledFuture scheduleSnapshotFuture; + public DatabaseTableMeta(){ } @@ -68,19 +77,10 @@ public class DatabaseTableMeta implements TableMetaTSDB { public boolean init(final String destination) { this.destination = destination; this.memoryTableMeta = new MemoryTableMeta(); - this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { - - @Override - public Thread newThread(Runnable r) { - Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]"); - thread.setDaemon(true); - return thread; - } - }); // 24小时生成一份snapshot if (snapshotInterval > 0) { - scheduler.scheduleWithFixedDelay(new Runnable() { + scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() { @Override public void run() { @@ -105,6 +105,26 @@ public class DatabaseTableMeta implements TableMetaTSDB { } return true; } + + @Override + public void destory() { + if (memoryTableMeta != null) { + memoryTableMeta.destory(); + } + + if (connection != null) { + try { + connection.disconnect(); + } catch (IOException e) { + logger.error("ERROR # disconnect meta connection for address:{}", connection.getConnector() + .getAddress(), e); + } + } + + if (scheduleSnapshotFuture != null) { + scheduleSnapshotFuture.cancel(false); + } + } @Override public TableMeta find(String schema, String table) { 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 904b55a9..c53e1a08 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 @@ -58,6 +58,11 @@ public class MemoryTableMeta implements TableMetaTSDB { public boolean init(String destination) { return true; } + + @Override + public void destory() { + tableMetas.clear(); + } public boolean apply(EntryPosition position, String schema, String ddl, String extra) { tableMetas.clear(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java index d712adbe..f968ae12 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java @@ -18,6 +18,11 @@ public interface TableMetaTSDB { */ public boolean init(String destination); + /** + * 销毁资源 + */ + public void destory(); + /** * 获取当前的表结构 */ diff --git a/parse/src/test/resources/tsdb/h2-tsdb.xml b/parse/src/test/resources/tsdb/h2-tsdb.xml index c0a3d385..dcb6bc47 100644 --- a/parse/src/test/resources/tsdb/h2-tsdb.xml +++ b/parse/src/test/resources/tsdb/h2-tsdb.xml @@ -7,7 +7,7 @@ default-autowire="byName"> - + diff --git a/parse/src/test/resources/tsdb/mysql-tsdb.xml b/parse/src/test/resources/tsdb/mysql-tsdb.xml index 099c034d..44f97203 100644 --- a/parse/src/test/resources/tsdb/mysql-tsdb.xml +++ b/parse/src/test/resources/tsdb/mysql-tsdb.xml @@ -7,7 +7,7 @@ default-autowire="byName"> - + diff --git a/pom.xml b/pom.xml index 5cb51eac..0bb4bc11 100644 --- a/pom.xml +++ b/pom.xml @@ -371,6 +371,7 @@ + + --> src/main/java src/test/java diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index 654a4b6b..8576835f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,21 +1,20 @@ package com.alibaba.otter.canal.kafka; -import java.util.List; -import java.util.Properties; - -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.Producer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.serializer.SerializerFeature; import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.spi.CanalMQProducer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.StringSerializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Properties; /** * kafka producer 主操作类 @@ -93,6 +92,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } else { // 发送扁平数据json @@ -110,6 +110,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } else { if (canalDestination.getPartitionHash() != null @@ -131,6 +132,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } } @@ -145,6 +147,7 @@ public class CanalKafkaProducer implements CanalMQProducer { logger.error(e.getMessage(), e); // producer.abortTransaction(); callback.rollback(); + return; } } } diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index 6173cebd..d8bd45e4 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -1,7 +1,11 @@ package com.alibaba.otter.canal.rocketmq; -import java.util.List; - +import com.alibaba.fastjson.JSON; +import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMQProducer; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.producer.DefaultMQProducer; @@ -12,12 +16,7 @@ import org.apache.rocketmq.remoting.exception.RemotingException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.alibaba.fastjson.JSON; -import com.alibaba.otter.canal.common.CanalMessageSerializer; -import com.alibaba.otter.canal.common.MQProperties; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.server.exception.CanalServerException; -import com.alibaba.otter.canal.spi.CanalMQProducer; +import java.util.List; public class CanalRocketMQProducer implements CanalMQProducer { @@ -67,6 +66,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) { logger.error("Send message error!", e); callback.rollback(); + return; } } else { List flatMessages = FlatMessage.messageConverter(data); @@ -90,6 +90,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (Exception e) { logger.error("send flat message to fixed partition error", e); callback.rollback(); + return; } } else { if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { @@ -124,6 +125,7 @@ public class CanalRocketMQProducer implements CanalMQProducer { } catch (Exception e) { logger.error("send flat message to hashed partition error", e); callback.rollback(); + return; } } }