diff --git a/README.md b/README.md
index b00d4a58..f3e29051 100644
--- a/README.md
+++ b/README.md
@@ -85,6 +85,7 @@ canal 特别设计了 client-server 模式,交互协议使用 protobuf 3.0 , c
- canal php客户端: [https://github.com/xingwenge/canal-php](https://github.com/xingwenge/canal-php)
- canal Python客户端:[https://github.com/haozi3156666/canal-python](https://github.com/haozi3156666/canal-python)
- canal Rust客户端:[https://github.com/laohanlinux/canal-rs](https://github.com/laohanlinux/canal-rs)
+- canal Nodejs客户端:[https://github.com/marmot-z/canal-nodejs](https://github.com/marmot-z/canal-nodejs)
canal 作为 MySQL binlog 增量获取和解析工具,可将变更记录投递到 MQ 系统中,比如 Kafka/RocketMQ,可以借助于 MQ 的多语言能力
diff --git a/admin/admin-ui/src/views/canalServer/CanalInstanceUpdate.vue b/admin/admin-ui/src/views/canalServer/CanalInstanceUpdate.vue
index cdb168f2..814ff44c 100644
--- a/admin/admin-ui/src/views/canalServer/CanalInstanceUpdate.vue
+++ b/admin/admin-ui/src/views/canalServer/CanalInstanceUpdate.vue
@@ -59,7 +59,7 @@ export default {
canalInstanceDetail(this.$route.query.id).then(response => {
const data = response.data
this.form.id = data.id
- this.form.name = data.name + '/instance.propertios'
+ this.form.name = data.name + '/instance.properties'
this.form.content = data.content
this.form.clusterServerId = data.clusterServerId
})
diff --git a/admin/admin-web/src/main/bin/startup.sh b/admin/admin-web/src/main/bin/startup.sh
index 2a6de188..f631cbbe 100644
--- a/admin/admin-web/src/main/bin/startup.sh
+++ b/admin/admin-web/src/main/bin/startup.sh
@@ -69,9 +69,12 @@ else
fi
if [ -n "$str" ]; then
- # JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
- # For G1
- JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ if [ $JavaVersion -ge 11 ] ; then
+ # For G1
+ JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ else
+ JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+ fi
else
JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"
fi
diff --git a/admin/admin-web/src/main/resources/canal-template.properties b/admin/admin-web/src/main/resources/canal-template.properties
index 9655f1e2..477bb957 100644
--- a/admin/admin-web/src/main/resources/canal-template.properties
+++ b/admin/admin-web/src/main/resources/canal-template.properties
@@ -25,7 +25,7 @@ canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
-# tcp, kafka, rocketMQ, rabbitMQ
+# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir}
@@ -96,7 +96,7 @@ canal.instance.tsdb.snapshot.expire = 360
#################################################
######### destinations #############
#################################################
-canal.destinations =
+canal.destinations =
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
@@ -148,8 +148,13 @@ kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0
kafka.kerberos.enable = false
-kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
-kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
+kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
+kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf
+
+# sasl demo
+# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
+# kafka.sasl.mechanism = SCRAM-SHA-512
+# kafka.security.protocol = SASL_PLAINTEXT
##################################################
######### RocketMQ #############
@@ -161,7 +166,7 @@ rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
-rocketmq.tag =
+rocketmq.tag =
##################################################
######### RabbitMQ #############
@@ -171,4 +176,12 @@ rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
-rabbitmq.deliveryMode =
\ No newline at end of file
+rabbitmq.deliveryMode =
+
+
+##################################################
+######### Pulsar #############
+##################################################
+pulsarmq.serverUrl =
+pulsarmq.roleToken =
+pulsarmq.topicTenantPrefix =
diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml
index af213c24..21213cf0 100644
--- a/client-adapter/common/pom.xml
+++ b/client-adapter/common/pom.xml
@@ -23,29 +23,8 @@
${project.version}
- org.springframework
- spring-core
-
-
- org.springframework
- spring-aop
-
-
- org.springframework
- spring-context
-
-
- org.springframework
- spring-jdbc
-
-
- org.springframework
- spring-orm
-
-
- org.springframework
- spring-test
- test
+ org.springframework.boot
+ spring-bootjoda-time
@@ -55,10 +34,6 @@
com.alibabadruid
-
- org.yaml
- snakeyaml
- org.powermockpowermock-api-mockito
diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
deleted file mode 100644
index 59a48d61..00000000
--- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/YmlConfigBinder.java
+++ /dev/null
@@ -1,155 +0,0 @@
-package com.alibaba.otter.canal.client.adapter.config;
-
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.Properties;
-
-import org.springframework.util.PropertyPlaceholderHelper;
-import org.springframework.util.StringUtils;
-
-import com.alibaba.otter.canal.client.adapter.config.bind.PropertiesConfigurationFactory;
-import com.alibaba.otter.canal.client.adapter.config.common.*;
-
-/**
- * 将yaml内容绑定到指定对象, 遵循spring yml的绑定规范
- *
- * @author reweerma 2019-2-1 上午9:14:02
- * @version 1.0.0
- */
-public class YmlConfigBinder {
-
- /**
- * 将当前内容绑定到指定对象
- *
- * @param content yml内容
- * @param clazz 指定对象类型
- * @return 对象
- */
- public static T bindYmlToObj(String content, Class clazz) {
- return bindYmlToObj(null, content, clazz, null);
- }
-
- /**
- * 将当前内容绑定到指定对象并指定内容编码格式
- *
- * @param content yml内容
- * @param clazz 指定对象类型
- * @param charset yml内容编码格式
- * @return 对象
- */
- public static T bindYmlToObj(String content, Class clazz, String charset) {
- return bindYmlToObj(null, content, clazz, charset);
- }
-
- /**
- * 将当前内容指定前缀部分绑定到指定对象
- *
- * @param prefix 指定前缀
- * @param content yml内容
- * @param clazz 指定对象类型
- * @return 对象
- */
- public static T bindYmlToObj(String prefix, String content, Class clazz) {
- return bindYmlToObj(prefix, content, clazz, null);
- }
-
- /**
- * 将当前内容指定前缀部分绑定到指定对象并指定内容编码格式
- *
- * @param prefix 指定前缀
- * @param content yml内容
- * @param clazz 指定对象类型
- * @param charset yml内容编码格式
- * @return 对象
- */
- public static T bindYmlToObj(String prefix, String content, Class clazz, String charset) {
- return bindYmlToObj(prefix, content, clazz, charset, null);
- }
-
- /**
- * 将当前内容指定前缀部分绑定到指定对象并用环境变量中的属性替换占位符, 例: 当前内容有属性 zkServers: ${zookeeper.servers}
- * 在envProperties中有属性 zookeeper.servers:
- * 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 则当前内容 zkServers 会被替换为
- * zkServers: 192.168.0.1:2181,192.168.0.1:2181,192.168.0.1:2181 注: 假设绑定的类中
- * zkServers 属性是 List 对象, 则会自动映射成List
- *
- * @param prefix 指定前缀
- * @param content yml内容
- * @param clazz 指定对象类型
- * @param charset yml内容编码格式
- * @return 对象
- */
- public static T bindYmlToObj(String prefix, String content, Class clazz, String charset,
- Properties baseProperties) {
- try {
- byte[] contentBytes;
- if (charset == null) {
- contentBytes = content.getBytes("UTF-8");
- } else {
- contentBytes = content.getBytes(charset);
- }
- YamlPropertySourceLoader propertySourceLoader = new YamlPropertySourceLoader();
- Resource configResource = new ByteArrayResource(contentBytes);
- PropertySource> propertySource = propertySourceLoader.load("manualBindConfig", configResource, null);
-
- if (propertySource == null) {
- return null;
- }
-
- Properties properties = new Properties();
- Map propertiesRes = new LinkedHashMap<>();
- if (!StringUtils.isEmpty(prefix) && !prefix.endsWith(".")) {
- prefix = prefix + ".";
- }
-
- properties.putAll((Map, ?>) propertySource.getSource());
-
- if (baseProperties != null) {
- baseProperties.putAll(properties);
- properties = baseProperties;
- }
-
- for (Map.Entry, ?> entry : ((Map, ?>) propertySource.getSource()).entrySet()) {
- String key = (String) entry.getKey();
- Object value = entry.getValue();
-
- if (prefix != null) {
- if (key != null && key.startsWith(prefix)) {
- key = key.substring(prefix.length());
- } else {
- continue;
- }
- }
-
- if (value instanceof String && ((String) value).contains("${") && ((String) value).contains("}")) {
- PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("${", "}");
- value = propertyPlaceholderHelper.replacePlaceholders((String) value, properties);
- }
-
- propertiesRes.put(key, value);
- }
-
- if (propertiesRes.isEmpty()) {
- return null;
- }
-
- propertySource = new MapPropertySource(propertySource.getName(), propertiesRes);
-
- T target = clazz.newInstance();
-
- MutablePropertySources propertySources = new MutablePropertySources();
- propertySources.addFirst(propertySource);
-
- PropertiesConfigurationFactory
- org.springframework
- spring-core
- 5.0.5.RELEASE
-
-
- org.springframework
- spring-aop
- 5.0.5.RELEASE
-
-
- org.springframework
- spring-context
- 5.0.5.RELEASE
-
-
- org.springframework
- spring-jdbc
- 5.0.5.RELEASE
-
-
- org.springframework
- spring-orm
- 5.0.5.RELEASE
-
-
- org.springframework
- spring-test
- 5.0.5.RELEASE
- test
+ org.springframework.boot
+ spring-boot
+ 2.0.1.RELEASEcom.h2database
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java
index d65e0ec5..cf57c75d 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/config/ConfigLoader.java
@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.client.adapter.rdb.config;
+import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
@@ -7,7 +8,6 @@ import java.util.Properties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
/**
@@ -32,8 +32,7 @@ public class ConfigLoader {
Map configContentMap = MappingConfigsLoader.loadConfigs("rdb");
configContentMap.forEach((fileName, content) -> {
- MappingConfig config = YmlConfigBinder
- .bindYmlToObj(null, content, MappingConfig.class, null, envProperties);
+ MappingConfig config = YamlUtils.ymlToObj(null, content, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
index f726ab69..794b021a 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/monitor/RdbConfigMonitor.java
@@ -1,10 +1,10 @@
package com.alibaba.otter.canal.client.adapter.rdb.monitor;
-import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.rdb.RdbAdapter;
import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
import com.alibaba.otter.canal.client.adapter.support.Util;
+import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import java.io.File;
import java.util.Properties;
import org.apache.commons.io.filefilter.FileFilterUtils;
@@ -35,7 +35,7 @@ public class RdbConfigMonitor {
File confDir = Util.getConfDirPath(adapterName);
try {
FileAlterationObserver observer = new FileAlterationObserver(confDir,
- FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
+ FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter("yml")));
FileListener listener = new FileListener();
observer.addListener(listener);
fileMonitor = new FileAlterationMonitor(3000, observer);
@@ -62,16 +62,15 @@ public class RdbConfigMonitor {
try {
// 加载新增的配置文件
String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
- MappingConfig config = YmlConfigBinder
- .bindYmlToObj(null, configContent, MappingConfig.class, null, envProperties);
+ MappingConfig config = YamlUtils
+ .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
config.validate();
boolean result = rdbAdapter.addConfig(file.getName(), config);
if (result) {
- logger.info("Add a new rdb mapping config: {} to canal adapter",
- file.getName());
+ logger.info("Add a new rdb mapping config: {} to canal adapter", file.getName());
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
@@ -85,14 +84,14 @@ public class RdbConfigMonitor {
try {
if (rdbAdapter.getRdbMapping().containsKey(file.getName())) {
// 加载配置文件
- String configContent = MappingConfigsLoader.loadConfig(adapterName + File.separator + file.getName());
+ String configContent = MappingConfigsLoader
+ .loadConfig(adapterName + File.separator + file.getName());
if (configContent == null) {
onFileDelete(file);
return;
}
- MappingConfig config = YmlConfigBinder
- .bindYmlToObj(null, configContent, MappingConfig.class, null,
- envProperties);
+ MappingConfig config = YamlUtils
+ .ymlToObj(null, configContent, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
index d1a511bd..18a73404 100644
--- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
+++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java
@@ -58,6 +58,7 @@ public class RdbEtlService extends AbstractEtlService {
Map columnsMap = new LinkedHashMap<>();
Map columnType = new LinkedHashMap<>();
DruidDataSource dataSource = (DruidDataSource) srcDS;
+ String backtick = SyncUtil.getBacktickByDbType(dataSource.getDbType());
Util.sqlRS(targetDS,
"SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " LIMIT 1 ",
@@ -91,7 +92,7 @@ public class RdbEtlService extends AbstractEtlService {
.append(SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()))
.append(" (");
columnsMap
- .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(","));
+ .forEach((targetColumnName, srcColumnName) -> insertSql.append(backtick).append(targetColumnName).append(backtick).append(","));
int len = insertSql.length();
insertSql.delete(len - 1, len).append(") VALUES (");
@@ -101,6 +102,7 @@ public class RdbEtlService extends AbstractEtlService {
}
len = insertSql.length();
insertSql.delete(len - 1, len).append(")");
+ logger.info("executeSqlImport sql:{}",insertSql.toString());
try (Connection connTarget = targetDS.getConnection();
PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) {
connTarget.setAutoCommit(false);
@@ -114,7 +116,7 @@ public class RdbEtlService extends AbstractEtlService {
Map pkVal = new LinkedHashMap<>();
StringBuilder deleteSql = new StringBuilder(
"DELETE FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType()) + " WHERE ");
- appendCondition(dbMapping, deleteSql, pkVal, rs);
+ appendCondition(dbMapping, deleteSql, pkVal, rs, backtick);
try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) {
int k = 1;
for (Object val : pkVal.values()) {
@@ -132,7 +134,6 @@ public class RdbEtlService extends AbstractEtlService {
}
Integer type = columnType.get(targetClolumnName.toLowerCase());
-
Object value = rs.getObject(srcColumnName);
if (value != null) {
SyncUtil.setPStmt(type, pstmt, value, i);
@@ -180,7 +181,7 @@ public class RdbEtlService extends AbstractEtlService {
* 拼接目标表主键where条件
*/
private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values,
- ResultSet rs) throws SQLException {
+ ResultSet rs, String backtick) throws SQLException {
// 拼接主键
for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) {
String targetColumnName = entry.getKey();
@@ -188,7 +189,7 @@ public class RdbEtlService extends AbstractEtlService {
if (srcColumnName == null) {
srcColumnName = targetColumnName;
}
- sql.append(targetColumnName).append("=? AND ");
+ sql.append(backtick).append(targetColumnName).append(backtick).append("=? AND ");
values.put(targetColumnName, rs.getObject(srcColumnName));
}
int len = sql.length();
diff --git a/client-adapter/tablestore/src/main/java/com/alibaba/otter/canal/client/adapter/tablestore/config/ConfigLoader.java b/client-adapter/tablestore/src/main/java/com/alibaba/otter/canal/client/adapter/tablestore/config/ConfigLoader.java
index f064fdfa..19b4a8f0 100644
--- a/client-adapter/tablestore/src/main/java/com/alibaba/otter/canal/client/adapter/tablestore/config/ConfigLoader.java
+++ b/client-adapter/tablestore/src/main/java/com/alibaba/otter/canal/client/adapter/tablestore/config/ConfigLoader.java
@@ -1,7 +1,7 @@
package com.alibaba.otter.canal.client.adapter.tablestore.config;
-import com.alibaba.otter.canal.client.adapter.config.YmlConfigBinder;
import com.alibaba.otter.canal.client.adapter.support.MappingConfigsLoader;
+import com.alibaba.otter.canal.client.adapter.support.YamlUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
@@ -32,8 +32,7 @@ public class ConfigLoader {
Map configContentMap = MappingConfigsLoader.loadConfigs("tablestore");
configContentMap.forEach((fileName, content) -> {
- MappingConfig config = YmlConfigBinder
- .bindYmlToObj(null, content, MappingConfig.class, null, envProperties);
+ MappingConfig config = YamlUtils.ymlToObj(null, content, MappingConfig.class, null, envProperties);
if (config == null) {
return;
}
@@ -45,7 +44,8 @@ public class ConfigLoader {
result.put(fileName, config);
});
- logger.info("## Tablestore mapping config loaded:" + StringUtils.collectionToCommaDelimitedString(result.keySet()));
+ logger.info(
+ "## Tablestore mapping config loaded:" + StringUtils.collectionToCommaDelimitedString(result.keySet()));
return result;
}
}
diff --git a/common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java b/common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java
index fe983dc2..bc068abd 100644
--- a/common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java
+++ b/common/src/main/java/com/alibaba/otter/canal/common/zookeeper/running/ServerRunningMonitor.java
@@ -38,7 +38,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
private ServerRunningData serverData;
// 当前实际运行的节点状态信息
private volatile ServerRunningData activeData;
- private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1);
+ private ScheduledExecutorService delayExecutor;
private int delayTime = 5;
private ServerRunningListener listener;
@@ -73,7 +73,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
initRunning();
} else {
// 否则就是等待delayTime,避免因网络瞬端或者zk异常,导致出现频繁的切换操作
- delayExector.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS);
+ delayExecutor.schedule(() -> initRunning(), delayTime, TimeUnit.SECONDS);
}
}
@@ -90,6 +90,7 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
try {
processStart();
if (zkClient != null) {
+ delayExecutor = Executors.newScheduledThreadPool(1);
// 如果需要尽可能释放instance资源,不需要监听running节点,不然即使stop了这台机器,另一台机器立马会start
String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.subscribeDataChanges(path, dataListener);
@@ -122,6 +123,10 @@ public class ServerRunningMonitor extends AbstractCanalLifeCycle {
if (zkClient != null) {
String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.unsubscribeDataChanges(path, dataListener);
+ if (delayExecutor != null) {
+ delayExecutor.shutdown();
+ delayExecutor = null;
+ }
releaseRunning(); // 尝试一下release
} else {
diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java
index d95087c6..6d7bb283 100644
--- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java
+++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java
@@ -107,11 +107,11 @@ public class AviaterRegexFilter {
private List completionPattern(List patterns) {
List result = new ArrayList<>();
for (String pattern : patterns) {
- StringBuffer stringBuffer = new StringBuffer();
- stringBuffer.append("^");
- stringBuffer.append(pattern);
- stringBuffer.append("$");
- result.add(stringBuffer.toString());
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("^");
+ stringBuilder.append(pattern);
+ stringBuilder.append("$");
+ result.add(stringBuilder.toString());
}
return result;
}
diff --git a/connector/pom.xml b/connector/pom.xml
index 744c8e20..951da13d 100644
--- a/connector/pom.xml
+++ b/connector/pom.xml
@@ -39,7 +39,7 @@
central
- http://repo1.maven.org/maven2
+ https://repo1.maven.org/maven2true
@@ -49,7 +49,7 @@
java.net
- http://download.java.net/maven/2/
+ https://download.java.net/maven/2/true
@@ -59,7 +59,7 @@
aliyun
- http://maven.aliyun.com/nexus/content/groups/public/
+ https://maven.aliyun.com/nexus/content/groups/public/true
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/CharsetConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/CharsetConversion.java
index 725abe60..c81dcc6f 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/CharsetConversion.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/CharsetConversion.java
@@ -21,6 +21,7 @@ public final class CharsetConversion {
protected final String mysqlCharset;
protected final String mysqlCollation;
protected final String javaCharset;
+ protected final Charset charset;
Entry(final int id, String mysqlCharset, // NL
String mysqlCollation, String javaCharset){
@@ -28,6 +29,7 @@ public final class CharsetConversion {
this.mysqlCharset = mysqlCharset;
this.mysqlCollation = mysqlCollation;
this.javaCharset = javaCharset;
+ this.charset = Charset.isSupported(javaCharset) ? Charset.forName(javaCharset) : null;
}
}
@@ -390,6 +392,23 @@ public final class CharsetConversion {
}
}
+ public static Charset getNioCharset(final int id) {
+ Entry entry = getEntry(id);
+
+ if (entry != null) {
+ if (entry.charset != null) {
+ return entry.charset;
+ } else {
+ logger.warn("Unknown java charset for: id = " + id + ", name = " + entry.mysqlCharset + ", coll = "
+ + entry.mysqlCollation);
+ return null;
+ }
+ } else {
+ logger.warn("Unexpect mysql charset: " + id);
+ return null;
+ }
+ }
+
public static void main(String[] args) {
for (int i = 0; i < entries.length; i++) {
Entry entry = entries[i];
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java
index f04daa1d..bcfd26b6 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java
@@ -1,5 +1,7 @@
package com.taobao.tddl.dbsync.binlog;
+import java.nio.charset.Charset;
+
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr;
@@ -57,23 +59,27 @@ public class JsonConversion {
public static final int VALUE_ENTRY_SIZE_LARGE = (1 + LARGE_OFFSET_SIZE);
public static Json_Value parse_value(int type, LogBuffer buffer, long len, String charsetName) {
+ return parse_value(type, buffer, len, Charset.forName(charsetName));
+ }
+
+ public static Json_Value parse_value(int type, LogBuffer buffer, long len, Charset charset) {
buffer = buffer.duplicate(buffer.position(), (int) len);
switch (type) {
case JSONB_TYPE_SMALL_OBJECT:
- return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charsetName);
+ return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charset);
case JSONB_TYPE_LARGE_OBJECT:
- return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charsetName);
+ return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charset);
case JSONB_TYPE_SMALL_ARRAY:
- return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charsetName);
+ return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charset);
case JSONB_TYPE_LARGE_ARRAY:
- return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charsetName);
+ return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charset);
default:
- return parse_scalar(type, buffer, len, charsetName);
+ return parse_scalar(type, buffer, len, charset);
}
}
private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large,
- String charsetName) {
+ Charset charset) {
long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
if (len < 2 * offset_size) {
throw new IllegalArgumentException("illegal json data");
@@ -100,7 +106,7 @@ public class JsonConversion {
return large ? buffer.getUint32() : buffer.getUint16();
}
- private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, String charsetName) {
+ private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, Charset charset) {
switch (type) {
case JSONB_TYPE_LITERAL:
/* purecov: inspected */
@@ -156,7 +162,7 @@ public class JsonConversion {
}
return new Json_Value(Json_enum_type.STRING, buffer.rewind()
.forward((int) n)
- .getFixString((int) str_len, charsetName));
+ .getFixString((int) str_len, charset));
case JSONB_TYPE_OPAQUE:
/*
* There should always be at least one byte, which tells the
@@ -242,7 +248,7 @@ public class JsonConversion {
this.m_large = large;
}
- public String key(int i, String charsetName) {
+ public String key(int i, Charset charset) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -255,10 +261,10 @@ public class JsonConversion {
// entry, always two
// bytes.
long key_length = m_data.getUint16();
- return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charsetName);
+ return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charset);
}
- public Json_Value element(int i, String charsetName) {
+ public Json_Value element(int i, Charset charset) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -271,13 +277,13 @@ public class JsonConversion {
int type = m_data.forward(entry_offset).getUint8();
if (type == JSONB_TYPE_INT16 || type == JSONB_TYPE_UINT16 || type == JSONB_TYPE_LITERAL
|| (m_large && (type == JSONB_TYPE_INT32 || type == JSONB_TYPE_UINT32))) {
- return parse_scalar(type, m_data, value_entry_size - 1, charsetName);
+ return parse_scalar(type, m_data, value_entry_size - 1, charset);
}
int value_offset = (int) read_offset_or_size(m_data, m_large);
- return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charsetName);
+ return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charset);
}
- public StringBuilder toJsonString(StringBuilder buf, String charsetName) {
+ public StringBuilder toJsonString(StringBuilder buf, Charset charset) {
switch (m_type) {
case OBJECT:
buf.append("{");
@@ -285,9 +291,9 @@ public class JsonConversion {
if (i > 0) {
buf.append(", ");
}
- buf.append('"').append(key(i, charsetName)).append('"');
+ buf.append('"').append(key(i, charset)).append('"');
buf.append(": ");
- element(i, charsetName).toJsonString(buf, charsetName);
+ element(i, charset).toJsonString(buf, charset);
}
buf.append("}");
break;
@@ -297,7 +303,7 @@ public class JsonConversion {
if (i > 0) {
buf.append(", ");
}
- element(i, charsetName).toJsonString(buf, charsetName);
+ element(i, charset).toJsonString(buf, charset);
}
buf.append("]");
break;
@@ -398,7 +404,7 @@ public class JsonConversion {
}
buf.append('"').append(text).append('"');
} else {
- text = m_data.getFixString((int) m_length, charsetName);
+ text = m_data.getFixString((int) m_length, charset);
buf.append('"').append(escapse(text)).append('"');
}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
index 878a0751..eb8f6dde 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
@@ -1,5 +1,6 @@
package com.taobao.tddl.dbsync.binlog;
+import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
@@ -36,6 +37,11 @@ public class JsonDiffConversion {
public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
String charsetName) {
+ return print_json_diff(buffer, len, columnName, columnIndex, Charset.forName(charsetName));
+ }
+
+ public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
+ Charset charset) {
int position = buffer.position();
List operation_names = new ArrayList<>();
while (buffer.hasRemaining()) {
@@ -111,14 +117,14 @@ public class JsonDiffConversion {
Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
buffer,
value_length - 1,
- charsetName);
+ charset);
buffer.forward((int) value_length - 1);
// Read value
if (jsonValue.m_type == Json_enum_type.ERROR) {
throw new IllegalArgumentException("parsing json value");
}
StringBuilder jsonBuilder = new StringBuilder();
- jsonValue.toJsonString(jsonBuilder, charsetName);
+ jsonValue.toJsonString(jsonBuilder, charset);
builder.append(jsonBuilder);
}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
index 408eaa4e..a2594948 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
@@ -5,6 +5,8 @@ import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.BitSet;
@@ -16,6 +18,8 @@ import java.util.BitSet;
*/
public class LogBuffer {
+ static final BigDecimal DECIMAL_ZERO_1_SCALE = BigDecimal.valueOf(0, 1);
+ static final BigDecimal DECIMAL_ONE_1_SCALE = BigDecimal.valueOf(10, 1);
protected byte[] buffer;
protected int origin, limit;
@@ -1038,29 +1042,27 @@ public class LogBuffer {
}
}
- /* default ANSI charset */
- public static final String ISO_8859_1 = "ISO-8859-1";
-
/**
* Return fix length string from buffer.
*/
public final String getFixString(final int pos, final int len) {
- return getFixString(pos, len, ISO_8859_1);
+ return getFixString(pos, len, StandardCharsets.ISO_8859_1);
}
/**
* Return next fix length string from buffer.
*/
public final String getFixString(final int len) {
- return getFixString(len, ISO_8859_1);
+ return getFixString(len, StandardCharsets.ISO_8859_1);
}
/**
* Return fix length string from buffer.
*/
- public final String getFixString(final int pos, final int len, String charsetName) {
- if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
- + (pos < 0 ? pos : (pos + len)));
+ public final String getFixString(final int pos, final int len, Charset charset) {
+ if (pos + len > limit || pos < 0) {
+ throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
+ }
final int from = origin + pos;
final int end = from + len;
@@ -1069,19 +1071,16 @@ public class LogBuffer {
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
- try {
- return new String(buf, from, found - from, charsetName);
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
- }
+ return new String(buf, from, found - from, charset);
}
/**
* Return next fix length string from buffer.
*/
- public final String getFixString(final int len, String charsetName) {
- if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
- + (position + len - origin));
+ public final String getFixString(final int len, Charset charset) {
+ if (position + len > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
+ }
final int from = position;
final int end = from + len;
@@ -1090,97 +1089,141 @@ public class LogBuffer {
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
- try {
- String string = new String(buf, from, found - from, charsetName);
- position += len;
- return string;
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+ String string = new String(buf, from, found - from, charset);
+ position += len;
+ return string;
+ }
+
+ public final String getFixName(final int len, Charset charset) {
+ if (position + len > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
+
+ final int from = position;
+ final int end = from + len;
+ byte[] buf = buffer;
+ int found = from;
+ for (; (found < end) && buf[found] != '\0'; found++)
+ /* empty loop */;
+
+ int length = found - from;
+ String string = null;
+ if (length <= 16) {
+ string = NameCache.name(buf, from, length, charset);
+ }
+ if (string == null) {
+ string = new String(buf, from, length, charset);
+ }
+ position += len;
+ return string;
+ }
+
+ public final String getFixName(final int len) {
+ if (position + len > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
+ }
+
+ final int from = position;
+ final int end = from + len;
+ byte[] buf = buffer;
+ int found = from;
+ for (; (found < end) && buf[found] != '\0'; found++)
+ /* empty loop */;
+
+ int length = found - from;
+ String string = null;
+ if (length <= 16) {
+ string = NameCache.name(buf, from, length, StandardCharsets.ISO_8859_1);
+ }
+ if (string == null) {
+ string = new String(buf, from, length, StandardCharsets.ISO_8859_1);
+ }
+ position += len;
+ return string;
}
/**
- * Return fix-length string from buffer without null-terminate checking. Fix
- * bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
+ * Return fix-length string from buffer without null-terminate checking.
+ * Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
*/
- public final String getFullString(final int pos, final int len, String charsetName) {
- if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
- + (pos < 0 ? pos : (pos + len)));
-
- try {
- return new String(buffer, origin + pos, len, charsetName);
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+ public final String getFullString(final int pos, final int len, Charset charset) {
+ if (pos + len > limit || pos < 0) {
+ throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
}
+
+ return new String(buffer, origin + pos, len, charset);
}
/**
* Return next fix-length string from buffer without null-terminate
- * checking. Fix bug #17 {@link https
- * ://github.com/AlibabaTech/canal/issues/17 }
+ * checking.
+ * Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
*/
- public final String getFullString(final int len, String charsetName) {
- if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
- + (position + len - origin));
-
- try {
- String string = new String(buffer, position, len, charsetName);
- position += len;
- return string;
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+ public final String getFullString(final int len, Charset charset) {
+ if (position + len > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
+
+ String string = new String(buffer, position, len, charset);
+ position += len;
+ return string;
}
+
/**
* Return dynamic length string from buffer.
*/
public final String getString(final int pos) {
- return getString(pos, ISO_8859_1);
+ if (pos >= limit || pos < 0) {
+ throw new IllegalArgumentException("limit excceed: " + pos);
+ }
+
+ byte[] buf = buffer;
+ final int len = (0xff & buf[origin + pos]);
+ if (pos + len + 1 > limit) {
+ throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
+ }
+
+ return new String(buf, origin + pos + 1, len, StandardCharsets.ISO_8859_1);
}
/**
* Return next dynamic length string from buffer.
*/
public final String getString() {
- return getString(ISO_8859_1);
- }
-
- /**
- * Return dynamic length string from buffer.
- */
- public final String getString(final int pos, String charsetName) {
- if (pos >= limit || pos < 0) throw new IllegalArgumentException("limit excceed: " + pos);
-
- byte[] buf = buffer;
- final int len = (0xff & buf[origin + pos]);
- if (pos + len + 1 > limit) throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
-
- try {
- return new String(buf, origin + pos + 1, len, charsetName);
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+ if (position >= origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + position);
}
- }
-
- /**
- * Return next dynamic length string from buffer.
- */
- public final String getString(String charsetName) {
- if (position >= origin + limit) throw new IllegalArgumentException("limit excceed: " + position);
byte[] buf = buffer;
final int len = (0xff & buf[position]);
- if (position + len + 1 > origin + limit) throw new IllegalArgumentException("limit excceed: "
- + (position + len + 1 - origin));
-
- try {
- String string = new String(buf, position + 1, len, charsetName);
- position += len + 1;
- return string;
- } catch (UnsupportedEncodingException e) {
- throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
+ if (position + len + 1 > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: "
+ + (position + len + 1 - origin));
}
+
+ String string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+ position += len + 1;
+ return string;
+ }
+
+ public final String getName() {
+ if (position >= origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + position);
+ }
+
+ byte[] buf = buffer;
+ final int len = (0xff & buf[position]);
+ if (position + len + 1 > origin + limit) {
+ throw new IllegalArgumentException("limit excceed: " + (position + len + 1 - origin));
+ }
+
+ String string = NameCache.name(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+ if (string == null) {
+ string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
+ }
+ position += len + 1;
+ return string;
}
/**
@@ -1263,6 +1306,15 @@ public class LogBuffer {
throw new IllegalArgumentException("limit excceed: " + (position + binSize - origin));
}
+ if (precision <= 20 && intg <= 18 && frac <= 9) {
+ // use compact BigDecimal
+ BigDecimal compactDecimal = getBigDecimalCompact(precision, scale, intg, frac, intg0x);
+ if (compactDecimal != null) {
+ position += binSize;
+ return compactDecimal;
+ }
+ }
+
BigDecimal decimal = getDecimal0(position, intg, frac, // NL
intg0,
frac0,
@@ -1272,6 +1324,162 @@ public class LogBuffer {
return decimal;
}
+ /**
+ * use compact BigDecimal
+ */
+ private BigDecimal getBigDecimalCompact(int precision, int scale, int intg, int frac, int intg0x) {
+ if (precision > 20 || intg > 18 && frac > 9) {
+ return null;
+ }
+
+ BigDecimal compactDecimal = null;
+ long longValue = 0;
+ int small = 0;
+ int pos = position;
+ byte b0 = buffer[position];
+ final int mask = ((b0 & 0x80) == 0x80) ? 0 : -1;
+ b0 ^= 0x80;
+
+ switch (intg0x) {
+ case 0:
+ if (intg == 0) {
+ longValue = 0;
+ } else {
+ longValue = (b0 << 24)
+ | ((0xff & buffer[pos + 1]) << 16)
+ | ((0xff & buffer[pos + 2]) << 8)
+ | (0xff & buffer[pos + 3]);
+ longValue ^= mask;
+ pos += 4;
+ }
+ break;
+ case 1:
+ case 2:
+ longValue = b0;
+ longValue ^= mask;
+ pos += 1;
+ break;
+ case 3:
+ case 4:
+ longValue = (b0 << 8) | (0xff & buffer[pos + 1]);
+ longValue ^= mask;
+ pos += 2;
+ break;
+ case 5:
+ case 6:
+ longValue = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
+ longValue ^= mask;
+ pos += 3;
+ break;
+ case 7:
+ case 8:
+ longValue = (b0 << 24)
+ | ((0xff & buffer[pos + 1]) << 16)
+ | ((0xff & buffer[pos + 2]) << 8)
+ | (0xff & buffer[pos + 3]);
+ longValue ^= mask;
+ pos += 4;
+ break;
+ default:
+ break;
+ }
+
+ if (intg > 9) {
+ int value = getInt32BE(buffer, pos);
+ value ^= mask;
+ longValue *= 1_000_000_000;
+ longValue += value;
+ pos += 4;
+ }
+
+ int power = 1;
+ if (pos != position) {
+ b0 = buffer[pos];
+ }
+ switch (frac) {
+ case 0:
+ small = 0;
+ break;
+ case 1:
+ case 2:
+ power = frac == 1 ? 10 : 100;
+ small = b0;
+ small ^= mask;
+ break;
+ case 3:
+ case 4:
+ power = frac == 3 ? 1_000 : 10_000;
+ small = (b0 << 8) | (0xff & buffer[pos + 1]);
+ small ^= mask;
+ break;
+ case 5:
+ case 6:
+ power = frac == 5 ? 100_000 : 1_000_000;
+ small = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
+ small ^= mask;
+ break;
+ case 7:
+ case 8:
+ case 9:
+ power = frac == 7 ? 10_000_000 : frac == 8 ? 100_000_000 : 1_000_000_000;
+ small = (b0 << 24)
+ | ((0xff & buffer[pos + 1]) << 16)
+ | ((0xff & buffer[pos + 2]) << 8)
+ | (0xff & buffer[pos + 3]);
+ small ^= mask;
+ break;
+ default:
+ break;
+ }
+
+ if (small == 0) {
+ if (frac == 0) {
+ if (mask != 0) {
+ longValue = -longValue;
+ }
+ compactDecimal = BigDecimal.valueOf(longValue, 0);
+ } else if (longValue == 0) {
+ if (scale < 9) {
+ compactDecimal = DECIMAL_ZERO_1_SCALE;
+ } else {
+ compactDecimal = BigDecimal.valueOf(longValue, scale);
+ }
+ } else {
+ if (scale < 9) {
+ long unscaleValue = longValue * 10;
+ if (unscaleValue >= longValue) {
+ if (mask != 0) {
+ unscaleValue = -unscaleValue;
+ }
+ if (unscaleValue == 10) {
+ compactDecimal = DECIMAL_ONE_1_SCALE;
+ } else {
+ compactDecimal = BigDecimal.valueOf(unscaleValue, 1);
+ }
+ }
+ } else {
+ long unscaleValue = longValue * power;
+ if (unscaleValue >= longValue) {
+ if (mask != 0) {
+ unscaleValue = -unscaleValue;
+ }
+ compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
+ }
+ }
+ }
+ } else {
+ long unscaleValue = longValue * power + small;
+ if (unscaleValue >= longValue) {
+ if (mask != 0) {
+ unscaleValue = -unscaleValue;
+ }
+ compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
+ }
+ }
+
+ return compactDecimal;
+ }
+
/**
* Return big decimal from buffer.
*
@@ -1445,8 +1653,9 @@ public class LogBuffer {
}
d_copy[begin] ^= 0x80; /* restore sign */
- String decimal = String.valueOf(buf, 0, pos);
- return new BigDecimal(decimal);
+ // String decimal = String.valueOf(buf, 0, pos);
+ // return new BigDecimal(decimal);
+ return new BigDecimal(buf, 0, pos);
}
/**
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java
index dbb19701..af237e27 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java
@@ -197,8 +197,9 @@ public final class LogDecoder {
}
case LogEvent.SLAVE_EVENT: /* can never happen (unused event) */
{
- if (logger.isWarnEnabled()) logger.warn("Skipping unsupported SLAVE_EVENT from: "
- + context.getLogPosition());
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported SLAVE_EVENT from: " + context.getLogPosition());
+ }
break;
}
case LogEvent.CREATE_FILE_EVENT: {
@@ -264,22 +265,25 @@ public final class LogDecoder {
return descriptionEvent;
}
case LogEvent.PRE_GA_WRITE_ROWS_EVENT: {
- if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: "
- + context.getLogPosition());
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: " + context.getLogPosition());
+ }
// ev = new Write_rows_log_event_old(buf, event_len,
// description_event);
break;
}
case LogEvent.PRE_GA_UPDATE_ROWS_EVENT: {
- if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: "
- + context.getLogPosition());
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: " + context.getLogPosition());
+ }
// ev = new Update_rows_log_event_old(buf, event_len,
// description_event);
break;
}
case LogEvent.PRE_GA_DELETE_ROWS_EVENT: {
- if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: "
- + context.getLogPosition());
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: " + context.getLogPosition());
+ }
// ev = new Delete_rows_log_event_old(buf, event_len,
// description_event);
break;
@@ -356,10 +360,16 @@ public final class LogDecoder {
return event;
}
case LogEvent.TRANSACTION_PAYLOAD_EVENT: {
- TransactionPayloadLogEvent event = new TransactionPayloadLogEvent(header, buffer, descriptionEvent);
- /* updating position in context */
- logPosition.position = header.getLogPos();
- return event;
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported MySQL TRANSACTION_PAYLOAD_EVENT from: " + context.getLogPosition());
+ }
+ break;
+
+ // TransactionPayloadLogEvent event = new TransactionPayloadLogEvent(header,
+ // buffer, descriptionEvent);
+ // /* updating position in context */
+ // logPosition.position = header.getLogPos();
+ // return event;
}
case LogEvent.VIEW_CHANGE_EVENT: {
ViewChangeEvent event = new ViewChangeEvent(header, buffer, descriptionEvent);
@@ -418,6 +428,39 @@ public final class LogDecoder {
logPosition.position = header.getLogPos();
return event;
}
+ case LogEvent.HEARTBEAT_LOG_EVENT_V2: {
+ HeartbeatV2LogEvent event = new HeartbeatV2LogEvent(header, buffer, descriptionEvent);
+ /* updating position in context */
+ logPosition.position = header.getLogPos();
+ return event;
+ }
+ case LogEvent.QUERY_COMPRESSED_EVENT: {
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported MaraiDB QUERY_COMPRESSED_EVENT from: " + context.getLogPosition());
+ }
+ break;
+ }
+ case LogEvent.WRITE_ROWS_COMPRESSED_EVENT_V1:
+ case LogEvent.WRITE_ROWS_COMPRESSED_EVENT: {
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported MaraiDB WRITE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
+ }
+ break;
+ }
+ case LogEvent.UPDATE_ROWS_COMPRESSED_EVENT_V1:
+ case LogEvent.UPDATE_ROWS_COMPRESSED_EVENT: {
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported MaraiDB UPDATE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
+ }
+ break;
+ }
+ case LogEvent.DELETE_ROWS_COMPRESSED_EVENT_V1:
+ case LogEvent.DELETE_ROWS_COMPRESSED_EVENT: {
+ if (logger.isWarnEnabled()) {
+ logger.warn("Skipping unsupported MaraiDB DELETE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
+ }
+ break;
+ }
default:
/*
* Create an object of Ignorable_log_event for unrecognized
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java
index 74e83c33..ace5f2fd 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java
@@ -178,7 +178,10 @@ public abstract class LogEvent {
/* mysql 8.0.20 */
public static final int TRANSACTION_PAYLOAD_EVENT = 40;
- public static final int MYSQL_ENUM_END_EVENT = 41;
+ /* mysql 8.0.26 */
+ public static final int HEARTBEAT_LOG_EVENT_V2 = 41;
+
+ public static final int MYSQL_ENUM_END_EVENT = 42;
// mariaDb 5.5.34
/* New MySQL/Sun events are to be added right above this comment */
@@ -210,8 +213,23 @@ public abstract class LogEvent {
public static final int START_ENCRYPTION_EVENT = 164;
+ // mariadb 10.10.1
+ /*
+ * Compressed binlog event. Note that the order between WRITE/UPDATE/DELETE
+ * events is significant; this is so that we can convert from the compressed to
+ * the uncompressed event type with (type-WRITE_ROWS_COMPRESSED_EVENT +
+ * WRITE_ROWS_EVENT) and similar for _V1.
+ */
+ public static final int QUERY_COMPRESSED_EVENT = 165;
+ public static final int WRITE_ROWS_COMPRESSED_EVENT_V1 = 166;
+ public static final int UPDATE_ROWS_COMPRESSED_EVENT_V1 = 167;
+ public static final int DELETE_ROWS_COMPRESSED_EVENT_V1 = 168;
+ public static final int WRITE_ROWS_COMPRESSED_EVENT = 169;
+ public static final int UPDATE_ROWS_COMPRESSED_EVENT = 170;
+ public static final int DELETE_ROWS_COMPRESSED_EVENT = 171;
+
/** end marker */
- public static final int ENUM_END_EVENT = 165;
+ public static final int ENUM_END_EVENT = 171;
/**
* 1 byte length, 1 byte format Length is total length in bytes, including 2
@@ -358,6 +376,7 @@ public abstract class LogEvent {
case INCIDENT_EVENT:
return "Incident";
case HEARTBEAT_LOG_EVENT:
+ case HEARTBEAT_LOG_EVENT_V2:
return "Heartbeat";
case IGNORABLE_LOG_EVENT:
return "Ignorable";
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/NameCache.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/NameCache.java
new file mode 100644
index 00000000..e10a8beb
--- /dev/null
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/NameCache.java
@@ -0,0 +1 @@
+package com.taobao.tddl.dbsync.binlog;
import java.nio.charset.Charset;
/**
* string cache
* @since 1.1.7
*/
public class NameCache {
static final NameCacheEntry[] NAME_CACHE = new NameCacheEntry[8192];
static final NameCacheEntry2[] NAME_CACHE2 = new NameCacheEntry2[8192];
static final class NameCacheEntry {
final String name;
final long value;
public NameCacheEntry(String name, long value){
this.name = name;
this.value = value;
}
}
static final class NameCacheEntry2 {
final String name;
final long value0;
final long value1;
public NameCacheEntry2(String name, long value0, long value1){
this.name = name;
this.value0 = value0;
this.value1 = value1;
}
}
static String name(byte[] buf, int from, int length, Charset charset) {
long nameValue0 = -1, nameValue1 = -1;
switch (length) {
case 1:
nameValue0 = buf[from];
break;
case 2:
nameValue0 = (buf[from + 1] << 8) + (buf[from] & 0xFF);
break;
case 3:
nameValue0 = (buf[from + 2] << 16) + ((buf[from + 1] & 0xFF) << 8) + (buf[from] & 0xFF);
break;
case 4:
nameValue0 = (buf[from + 3] << 24) + ((buf[from + 2] & 0xFF) << 16) + ((buf[from + 1] & 0xFF) << 8)
+ (buf[from] & 0xFF);
break;
case 5:
nameValue0 = (((long) buf[from + 4]) << 32) + ((buf[from + 3] & 0xFFL) << 24)
+ ((buf[from + 2] & 0xFFL) << 16) + ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
break;
case 6:
nameValue0 = (((long) buf[from + 5]) << 40) + ((buf[from + 4] & 0xFFL) << 32)
+ ((buf[from + 3] & 0xFFL) << 24) + ((buf[from + 2] & 0xFFL) << 16)
+ ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
break;
case 7:
nameValue0 = (((long) buf[from + 6]) << 48) + ((buf[from + 5] & 0xFFL) << 40)
+ ((buf[from + 4] & 0xFFL) << 32) + ((buf[from + 3] & 0xFFL) << 24)
+ ((buf[from + 2] & 0xFFL) << 16) + ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
break;
case 8:
nameValue0 = (((long) buf[from + 7]) << 56) + ((buf[from + 6] & 0xFFL) << 48)
+ ((buf[from + 5] & 0xFFL) << 40) + ((buf[from + 4] & 0xFFL) << 32)
+ ((buf[from + 3] & 0xFFL) << 24) + ((buf[from + 2] & 0xFFL) << 16)
+ ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
break;
case 9:
nameValue0 = buf[from];
nameValue1 = (((long) buf[from + 8]) << 56) + ((buf[from + 7] & 0xFFL) << 48)
+ ((buf[from + 6] & 0xFFL) << 40) + ((buf[from + 5] & 0xFFL) << 32)
+ ((buf[from + 4] & 0xFFL) << 24) + ((buf[from + 3] & 0xFFL) << 16)
+ ((buf[from + 2] & 0xFFL) << 8) + (buf[from + 1] & 0xFFL);
break;
case 10:
nameValue0 = (buf[from + 1] << 8) + (buf[from]);
nameValue1 = (((long) buf[from + 9]) << 56) + ((buf[from + 8] & 0xFFL) << 48)
+ ((buf[from + 7] & 0xFFL) << 40) + ((buf[from + 6] & 0xFFL) << 32)
+ ((buf[from + 5] & 0xFFL) << 24) + ((buf[from + 4] & 0xFFL) << 16)
+ ((buf[from + 3] & 0xFFL) << 8) + (buf[from + 2] & 0xFFL);
break;
case 11:
nameValue0 = (buf[from + 2] << 16) + (buf[from + 1] << 8) + (buf[from]);
nameValue1 = (((long) buf[from + 10]) << 56) + ((buf[from + 9] & 0xFFL) << 48)
+ ((buf[from + 8] & 0xFFL) << 40) + ((buf[from + 7] & 0xFFL) << 32)
+ ((buf[from + 6] & 0xFFL) << 24) + ((buf[from + 5] & 0xFFL) << 16)
+ ((buf[from + 4] & 0xFFL) << 8) + (buf[from + 3] & 0xFFL);
break;
case 12:
nameValue0 = (buf[from + 3] << 24) + (buf[from + 2] << 16) + (buf[from + 1] << 8) + (buf[from]);
nameValue1 = (((long) buf[from + 11]) << 56) + ((buf[from + 10] & 0xFFL) << 48)
+ ((buf[from + 9] & 0xFFL) << 40) + ((buf[from + 8] & 0xFFL) << 32)
+ ((buf[from + 7] & 0xFFL) << 24) + ((buf[from + 6] & 0xFFL) << 16)
+ ((buf[from + 5] & 0xFFL) << 8) + (buf[from + 4] & 0xFFL);
break;
case 13:
nameValue0 = (((long) buf[from + 4]) << 32) + (((long) buf[from + 3]) << 24)
+ (((long) buf[from + 2]) << 16) + (((long) buf[from + 1]) << 8) + ((long) buf[from]);
nameValue1 = (((long) buf[from + 12]) << 56) + ((buf[from + 11] & 0xFFL) << 48)
+ ((buf[from + 10] & 0xFFL) << 40) + ((buf[from + 9] & 0xFFL) << 32)
+ ((buf[from + 8] & 0xFFL) << 24) + ((buf[from + 7] & 0xFFL) << 16)
+ ((buf[from + 6] & 0xFFL) << 8) + (buf[from + 5] & 0xFFL);
break;
case 14:
nameValue0 = (((long) buf[from + 5]) << 40) + ((buf[from + 4] & 0xFFL) << 32)
+ ((buf[from + 3] & 0xFFL) << 24) + ((buf[from + 2] & 0xFFL) << 16)
+ ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
nameValue1 = (((long) buf[from + 13]) << 56) + ((buf[from + 12] & 0xFFL) << 48)
+ ((buf[from + 11] & 0xFFL) << 40) + ((buf[from + 10] & 0xFFL) << 32)
+ ((buf[from + 9] & 0xFFL) << 24) + ((buf[from + 8] & 0xFFL) << 16)
+ ((buf[from + 7] & 0xFFL) << 8) + (buf[from + 6] & 0xFFL);
break;
case 15:
nameValue0 = (((long) buf[from + 6]) << 48) + ((buf[from + 5] & 0xFFL) << 40)
+ ((buf[from + 4] & 0xFFL) << 32) + ((buf[from + 3] & 0xFFL) << 24)
+ ((buf[from + 2] & 0xFFL) << 16) + ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
nameValue1 = (((long) buf[from + 14]) << 56) + ((buf[from + 13] & 0xFFL) << 48)
+ ((buf[from + 12] & 0xFFL) << 40) + ((buf[from + 11] & 0xFFL) << 32)
+ ((buf[from + 10] & 0xFFL) << 24) + ((buf[from + 9] & 0xFFL) << 16)
+ ((buf[from + 8] & 0xFFL) << 8) + (buf[from + 7] & 0xFFL);
break;
case 16:
nameValue0 = (((long) buf[from + 7]) << 56) + ((buf[from + 6] & 0xFFL) << 48)
+ ((buf[from + 5] & 0xFFL) << 40) + ((buf[from + 4] & 0xFFL) << 32)
+ ((buf[from + 3] & 0xFFL) << 24) + ((buf[from + 2] & 0xFFL) << 16)
+ ((buf[from + 1] & 0xFFL) << 8) + (buf[from] & 0xFFL);
nameValue1 = (((long) buf[from + 15]) << 56) + ((buf[from + 14] & 0xFFL) << 48)
+ ((buf[from + 13] & 0xFFL) << 40) + ((buf[from + 12] & 0xFFL) << 32)
+ ((buf[from + 11] & 0xFFL) << 24) + ((buf[from + 10] & 0xFFL) << 16)
+ ((buf[from + 9] & 0xFFL) << 8) + (buf[from + 8] & 0xFFL);
break;
default:
break;
}
if (nameValue0 != -1) {
if (nameValue1 != -1) {
int indexMask = ((int) nameValue1) & (NameCache.NAME_CACHE2.length - 1);
NameCache.NameCacheEntry2 entry = NameCache.NAME_CACHE2[indexMask];
if (entry == null) {
String name = new String(buf, from, length, charset);
NameCache.NAME_CACHE2[indexMask] = new NameCacheEntry2(name, nameValue0, nameValue1);
return name;
} else if (entry.value0 == nameValue0 && entry.value1 == nameValue1) {
return entry.name;
}
} else {
int indexMask = ((int) nameValue0) & (NAME_CACHE.length - 1);
NameCacheEntry entry = NAME_CACHE[indexMask];
if (entry == null) {
String name = new String(buf, from, length, charset);
NAME_CACHE[indexMask] = new NameCacheEntry(name, nameValue0);
return name;
} else if (entry.value == nameValue0) {
return entry.name;
}
}
}
return null;
}
}
\ No newline at end of file
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
index 49557fc6..3081eedc 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
@@ -18,6 +18,7 @@ public class GtidLogEvent extends LogEvent {
// / Length of SID in event encoding
public static final int ENCODED_SID_LENGTH = 16;
public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2;
+ public static final UUID UUID_ZERO = new UUID(0, 0);
private boolean commitFlag;
private UUID sid;
@@ -35,11 +36,13 @@ public class GtidLogEvent extends LogEvent {
buffer.position(commonHeaderLen);
commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH
- byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
- ByteBuffer bb = ByteBuffer.wrap(bs);
- long high = bb.getLong();
- long low = bb.getLong();
- sid = new UUID(high, low);
+ long high = buffer.getBeLong64();
+ long low = buffer.getBeLong64();
+ if (high == 0 && low == 0) {
+ sid = UUID_ZERO;
+ } else {
+ sid = new UUID(high, low);
+ }
gno = buffer.getLong64();
@@ -73,11 +76,11 @@ public class GtidLogEvent extends LogEvent {
return gno;
}
- public Long getLastCommitted() {
+ public long getLastCommitted() {
return lastCommitted;
}
- public Long getSequenceNumber() {
+ public long getSequenceNumber() {
return sequenceNumber;
}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatLogEvent.java
index f3948d7e..ea8c519e 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatLogEvent.java
@@ -3,6 +3,8 @@ package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
+import java.nio.charset.StandardCharsets;
+
/**
*
* Replication event to ensure to slave that master is alive.
@@ -35,7 +37,7 @@ public class HeartbeatLogEvent extends LogEvent {
identLen = FN_REFLEN - 1;
}
- logIdent = buffer.getFullString(commonHeaderLen, identLen, LogBuffer.ISO_8859_1);
+ logIdent = buffer.getFullString(commonHeaderLen, identLen, StandardCharsets.ISO_8859_1);
}
public int getIdentLen() {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java
new file mode 100644
index 00000000..e6619c7c
--- /dev/null
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java
@@ -0,0 +1,40 @@
+package com.taobao.tddl.dbsync.binlog.event;
+
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import com.taobao.tddl.dbsync.binlog.LogEvent;
+
+/**
+ *
+ * Replication event to ensure to replica that source is alive.
+ * The event is originated by source's dump thread and sent straight to
+ * replica without being logged. Slave itself does not store it in relay log
+ * but rather uses a data for immediate checks and throws away the event.
+ * Two members of the class m_log_filename and m_log_position comprise
+ * @see the rpl_event_coordinates instance. The coordinates that a heartbeat
+ * instance carries correspond to the last event source has sent from
+ * its binlog.
+ * Also this event will be generated only for the source server with
+ * version > 8.0.26
+ *
+ *
+ * @author jianghang 2022-09-01 下午16:36:29
+ * @version 1.1.6
+ * @since mysql 8.0.26
+ */
+public class HeartbeatV2LogEvent extends LogEvent {
+
+ private byte[] payload;
+
+ public HeartbeatV2LogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+ super(header);
+
+ final int commonHeaderLen = descriptionEvent.commonHeaderLen;
+ int payloadLenth = buffer.limit() - commonHeaderLen;
+ // see : https://github.com/mysql/mysql-server/commit/59e590738a772b74ad50c4a57c86aaa1bc6501c7#diff-184e9a7d8a58f080974e475d4199fe5c6da5518c8a2811cc5df5988c8f9e9797
+ payload = buffer.getData(payloadLenth);
+ }
+
+ public byte[] getPayload() {
+ return payload;
+ }
+}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/IncidentLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/IncidentLogEvent.java
index f0c03e56..f1a4b73d 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/IncidentLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/IncidentLogEvent.java
@@ -67,7 +67,7 @@ public final class IncidentLogEvent extends LogEvent {
incident = incidentNumber;
buffer.position(commonHeaderLen + postHeaderLen);
- message = buffer.getString();
+ message = buffer.getName();
}
public final int getIncident() {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java
index 8cc49795..b4fb5096 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java
@@ -433,7 +433,7 @@ public class QueryLogEvent extends LogEvent {
private int serverCollation = -1;
private int tvSec = -1;
private BigInteger ddlXid = BigInteger.valueOf(-1L);
- private String charsetName;
+ private Charset charset;
private String timezone;
@@ -498,12 +498,12 @@ public class QueryLogEvent extends LogEvent {
/* A 2nd variable part; this is common to all versions */
final int queryLen = dataLen - dbLen - 1;
- dbname = buffer.getFixString(dbLen + 1);
+ dbname = buffer.getFixName(dbLen + 1);
if (clientCharset >= 0) {
- charsetName = CharsetConversion.getJavaCharset(clientCharset);
+ charset = CharsetConversion.getNioCharset(clientCharset);
- if ((charsetName != null) && (Charset.isSupported(charsetName))) {
- query = buffer.getFixString(queryLen, charsetName);
+ if (charset != null) {
+ query = buffer.getFixString(queryLen, charset);
} else {
logger.warn("unsupported character set in query log: " + "\n ID = " + clientCharset + ", Charset = "
+ CharsetConversion.getCharset(clientCharset) + ", Collation = "
@@ -603,6 +603,13 @@ public class QueryLogEvent extends LogEvent {
*/
public static final int Q_HRNOW = 128;
+ /**
+ * Support MariaDB 10.10.1
+ */
+ public static final int Q_XID = 129;
+
+ public static final int Q_GTID_FLAGS3 = 130;
+
private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
int code = -1;
try {
@@ -613,10 +620,10 @@ public class QueryLogEvent extends LogEvent {
break;
case Q_SQL_MODE_CODE:
sql_mode = buffer.getLong64(); // QQ: Fix when sql_mode
- // is ulonglong
+ // is ulonglong
break;
case Q_CATALOG_NZ_CODE:
- catalog = buffer.getString();
+ catalog = buffer.getName();
break;
case Q_AUTO_INCREMENT:
autoIncrementIncrement = buffer.getUint16();
@@ -632,7 +639,7 @@ public class QueryLogEvent extends LogEvent {
serverCollation = buffer.getUint16();
break;
case Q_TIME_ZONE_CODE:
- timezone = buffer.getString();
+ timezone = buffer.getName();
break;
case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */
final int len = buffer.getUint8();
@@ -656,8 +663,8 @@ public class QueryLogEvent extends LogEvent {
buffer.forward(4);
break;
case Q_INVOKER:
- user = buffer.getString();
- host = buffer.getString();
+ user = buffer.getName();
+ host = buffer.getName();
break;
case Q_MICROSECONDS:
// when.tv_usec= uint3korr(pos);
@@ -678,7 +685,7 @@ public class QueryLogEvent extends LogEvent {
String mtsAccessedDbNames[] = new String[mtsAccessedDbs];
for (int i = 0; i < mtsAccessedDbs && buffer.position() < end; i++) {
int length = end - buffer.position();
- mtsAccessedDbNames[i] = buffer.getFixString(length < NAME_LEN ? length : NAME_LEN);
+ mtsAccessedDbNames[i] = buffer.getFixName(length < NAME_LEN ? length : NAME_LEN);
}
break;
case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
@@ -701,6 +708,25 @@ public class QueryLogEvent extends LogEvent {
// int when_sec_part = buffer.getUint24();
buffer.forward(3);
break;
+ case Q_XID:
+ // xid= uint8korr(pos);
+ buffer.forward(8);
+ break;
+ case Q_GTID_FLAGS3:
+ // gtid_flags_extra= *pos++;
+ // if (gtid_flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 |
+ // Gtid_log_event::FL_ROLLBACK_ALTER_E1)) {
+ // sa_seq_no = uint8korr(pos);
+ // pos+= 8;
+ // }
+
+ int gtid_flags_extra = buffer.getUint8();
+ final int FL_COMMIT_ALTER_E1= 4;
+ final int FL_ROLLBACK_ALTER_E1= 8;
+ if ((gtid_flags_extra & (FL_COMMIT_ALTER_E1 | FL_ROLLBACK_ALTER_E1))> 0) {
+ buffer.forward(8);
+ }
+ break;
default:
/*
* That's why you must write status vars in growing
@@ -740,16 +766,22 @@ public class QueryLogEvent extends LogEvent {
return "Q_TABLE_MAP_FOR_UPDATE_CODE";
case Q_MASTER_DATA_WRITTEN_CODE:
return "Q_MASTER_DATA_WRITTEN_CODE";
- case Q_UPDATED_DB_NAMES:
- return "Q_UPDATED_DB_NAMES";
+ case Q_INVOKER:
+ return "case Q_INVOKER";
case Q_MICROSECONDS:
return "Q_MICROSECONDS";
+ case Q_UPDATED_DB_NAMES:
+ return "Q_UPDATED_DB_NAMES";
+ case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
+ return "Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP";
case Q_DDL_LOGGED_WITH_XID:
return "Q_DDL_LOGGED_WITH_XID";
case Q_DEFAULT_COLLATION_FOR_UTF8MB4:
return "Q_DEFAULT_COLLATION_FOR_UTF8MB4";
case Q_SQL_REQUIRE_PRIMARY_KEY:
return "Q_SQL_REQUIRE_PRIMARY_KEY";
+ case Q_HRNOW:
+ return "Q_HRNOW";
}
return "CODE#" + code;
}
@@ -797,8 +829,8 @@ public class QueryLogEvent extends LogEvent {
return autoIncrementOffset;
}
- public final String getCharsetName() {
- return charsetName;
+ public final Charset getCharset() {
+ return charset;
}
public final String getTimezone() {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
index 30f05a45..79335e11 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
@@ -1,8 +1,11 @@
package com.taobao.tddl.dbsync.binlog.event;
import java.io.Serializable;
+import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.sql.Types;
+import java.time.ZoneId;
+import java.time.zone.ZoneRules;
import java.util.BitSet;
import org.apache.commons.logging.Log;
@@ -23,7 +26,11 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
*/
public final class RowsLogBuffer {
- protected static final Log logger = LogFactory.getLog(RowsLogBuffer.class);
+ protected static final Log logger = LogFactory.getLog(RowsLogBuffer.class);
+ public static final Long[] longCache = new Long[1024 * 128];
+ public static final int longCacheLimit = longCache.length + 127;
+ public static final Integer[] integerCache = new Integer[1024 * 128];
+ public static final int integerCacheLimit = longCache.length + 127;
public static final long DATETIMEF_INT_OFS = 0x8000000000L;
public static final long TIMEF_INT_OFS = 0x800000L;
@@ -33,7 +40,7 @@ public final class RowsLogBuffer {
private final LogBuffer buffer;
private final int columnLen;
private final int jsonColumnCount;
- private final String charsetName;
+ private final Charset charset;
private final BitSet nullBits;
private int nullBitIndex;
@@ -47,10 +54,10 @@ public final class RowsLogBuffer {
private int length;
private Serializable value;
- public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName, int jsonColumnCount, boolean partial){
+ public RowsLogBuffer(LogBuffer buffer, final int columnLen, Charset charset, int jsonColumnCount, boolean partial){
this.buffer = buffer;
this.columnLen = columnLen;
- this.charsetName = charsetName;
+ this.charset = charset;
this.partial = partial;
this.jsonColumnCount = jsonColumnCount;
this.nullBits = new BitSet(columnLen);
@@ -309,7 +316,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = unsigned ? Long.valueOf(buffer.getUint32()) :
// Integer.valueOf(buffer.getInt32());
- value = Integer.valueOf(buffer.getInt32());
+ value = valueOf(buffer.getInt32());
javaType = Types.INTEGER;
length = 4;
break;
@@ -318,7 +325,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint8() :
// buffer.getInt8());
- value = Integer.valueOf(buffer.getInt8());
+ value = valueOf(buffer.getInt8());
javaType = Types.TINYINT; // java.sql.Types.INTEGER;
length = 1;
break;
@@ -327,7 +334,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint16() :
// buffer.getInt16());
- value = Integer.valueOf((short) buffer.getInt16());
+ value = valueOf((short) buffer.getInt16());
javaType = Types.SMALLINT; // java.sql.Types.INTEGER;
length = 2;
break;
@@ -336,7 +343,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint24() :
// buffer.getInt24());
- value = Integer.valueOf(buffer.getInt24());
+ value = valueOf(buffer.getInt24());
javaType = Types.INTEGER;
length = 3;
break;
@@ -345,7 +352,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = unsigned ? buffer.getUlong64()) :
// Long.valueOf(buffer.getLong64());
- value = Long.valueOf(buffer.getLong64());
+ value = valueOf(buffer.getLong64());
javaType = Types.BIGINT; // Types.INTEGER;
length = 8;
break;
@@ -870,7 +877,7 @@ public final class RowsLogBuffer {
}
// logger.warn("MYSQL_TYPE_ENUM : This enumeration value is "
// + "only used internally and cannot exist in a binlog!");
- value = Integer.valueOf(int32);
+ value = valueOf(int32);
javaType = Types.INTEGER;
length = len;
break;
@@ -1017,7 +1024,7 @@ public final class RowsLogBuffer {
javaType = Types.VARBINARY;
value = binary;
} else {
- value = buffer.getFullString(len, charsetName);
+ value = buffer.getFullString(len, charset);
javaType = Types.VARCHAR;
}
@@ -1039,7 +1046,7 @@ public final class RowsLogBuffer {
javaType = Types.BINARY;
value = binary;
} else {
- value = buffer.getFullString(len, charsetName);
+ value = buffer.getFullString(len, charset);
javaType = Types.CHAR; // Types.VARCHAR;
}
length = len;
@@ -1074,7 +1081,7 @@ public final class RowsLogBuffer {
len,
columnName,
columnIndex,
- charsetName);
+ charset);
value = builder.toString();
buffer.position(position + len);
} else {
@@ -1088,9 +1095,9 @@ public final class RowsLogBuffer {
Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
buffer,
len - 1,
- charsetName);
+ charset);
StringBuilder builder = new StringBuilder();
- jsonValue.toJsonString(builder, charsetName);
+ jsonValue.toJsonString(builder, charset);
value = builder.toString();
buffer.position(position + len);
}
@@ -1211,4 +1218,38 @@ public final class RowsLogBuffer {
builder.append('0').append(digits[d]);
}
}
+
+ public static Long valueOf(long value) {
+ if (value >= -128 & value <= 127) {
+ // if (l >= -128 && l <= 127) {
+ return Long.valueOf(value);
+ }
+ if (value > 127 && value < longCacheLimit) {
+ int cacheIndex = (int) value - 127;
+ Long cacheValue = longCache[cacheIndex];
+ if (cacheValue == null) {
+ cacheValue = new Long(value);
+ longCache[cacheIndex] = cacheValue;
+ }
+ return cacheValue;
+ }
+ return new Long(value);
+ }
+
+ public static Integer valueOf(int value) {
+ if (value >= -128 & value <= 127) {
+ // if (l >= -128 && l <= 127) {
+ return Integer.valueOf(value);
+ }
+ if (value > 127 && value < integerCacheLimit) {
+ int cacheIndex = (int) value - 127;
+ Integer cacheValue = integerCache[cacheIndex];
+ if (cacheValue == null) {
+ cacheValue = new Integer(value);
+ integerCache[cacheIndex] = cacheValue;
+ }
+ return cacheValue;
+ }
+ return new Integer(value);
+ }
}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
index e1e78cd1..2683217b 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
@@ -1,5 +1,6 @@
package com.taobao.tddl.dbsync.binlog.event;
+import java.nio.charset.Charset;
import java.util.BitSet;
import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
@@ -219,8 +220,8 @@ public abstract class RowsLogEvent extends LogEvent {
return changeColumns;
}
- public final RowsLogBuffer getRowsBuf(String charsetName) {
- return new RowsLogBuffer(rowsBuf, columnLen, charsetName, jsonColumnCount, partial);
+ public final RowsLogBuffer getRowsBuf(Charset charset) {
+ return new RowsLogBuffer(rowsBuf, columnLen, charset, jsonColumnCount, partial);
}
public final int getFlags(final int flags) {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsQueryLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsQueryLogEvent.java
index 16293ae1..d4ba23c6 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsQueryLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsQueryLogEvent.java
@@ -2,6 +2,8 @@ package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
+import java.nio.charset.StandardCharsets;
+
/**
* @author jianghang 2013-4-8 上午12:36:29
* @version 1.0.3
@@ -23,7 +25,7 @@ public class RowsQueryLogEvent extends IgnorableLogEvent {
*/
int offset = commonHeaderLen + postHeaderLen + 1;
int len = buffer.limit() - offset;
- rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
+ rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
}
public String getRowsQuery() {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/StartLogEventV3.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/StartLogEventV3.java
index f230759b..37ed1ad9 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/StartLogEventV3.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/StartLogEventV3.java
@@ -36,7 +36,7 @@ public class StartLogEventV3 extends LogEvent {
buffer.position(descriptionEvent.commonHeaderLen);
binlogVersion = buffer.getUint16(); // ST_BINLOG_VER_OFFSET
- serverVersion = buffer.getFixString(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
+ serverVersion = buffer.getFixName(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
}
public StartLogEventV3(){
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
index 4f64a7d5..ead820ee 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
@@ -423,10 +423,10 @@ public final class TableMapLogEvent extends LogEvent {
/* Read the variable part of the event */
buffer.position(commonHeaderLen + postHeaderLen);
- dbname = buffer.getString();
+ dbname = buffer.getName();
buffer.forward(1); /* termination null */
// fixed issue #2714
- tblname = buffer.getString();
+ tblname = buffer.getName();
buffer.forward(1); /* termination null */
// Read column information from buffer
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UserVarLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UserVarLogEvent.java
index d6acb94c..19d5320e 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UserVarLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UserVarLogEvent.java
@@ -2,6 +2,7 @@ package com.taobao.tddl.dbsync.binlog.event;
import java.io.IOException;
import java.io.Serializable;
+import java.nio.charset.Charset;
import com.taobao.tddl.dbsync.binlog.CharsetConversion;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
@@ -100,8 +101,8 @@ public final class UserVarLogEvent extends LogEvent {
value = buffer.getDecimal(precision, scale); // bin2decimal
break;
case STRING_RESULT:
- String charsetName = CharsetConversion.getJavaCharset(charsetNumber);
- value = buffer.getFixString(valueLen, charsetName);
+ Charset charset = CharsetConversion.getNioCharset(charsetNumber);
+ value = buffer.getFixString(valueLen, charset);
break;
case ROW_RESULT:
// this seems to be banned in MySQL altogether
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/AnnotateRowsEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/AnnotateRowsEvent.java
index cbb83871..19992028 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/AnnotateRowsEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/AnnotateRowsEvent.java
@@ -5,6 +5,8 @@ import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import com.taobao.tddl.dbsync.binlog.event.IgnorableLogEvent;
import com.taobao.tddl.dbsync.binlog.event.LogHeader;
+import java.nio.charset.StandardCharsets;
+
/**
* mariadb的ANNOTATE_ROWS_EVENT类型
*
@@ -23,7 +25,7 @@ public class AnnotateRowsEvent extends IgnorableLogEvent {
int offset = commonHeaderLen + postHeaderLen;
int len = buffer.limit() - offset;
- rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
+ rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
}
public String getRowsQuery() {
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 27ecf3db..4980a292 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
@@ -52,7 +52,7 @@ public class BaseLogFetcherTest {
event.getHeader().getLogPos() - event.getHeader().getEventLen(),
event.getTable().getDbName(),
event.getTable().getTableName()));
- RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+ RowsLogBuffer buffer = event.getRowsBuf(charset);
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {
diff --git a/deployer/src/main/bin/startup.sh b/deployer/src/main/bin/startup.sh
index b8c0ed3e..583c080a 100644
--- a/deployer/src/main/bin/startup.sh
+++ b/deployer/src/main/bin/startup.sh
@@ -92,9 +92,12 @@ else
fi
if [ -n "$str" ]; then
- # JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
- # For G1
- JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ if [ $JavaVersion -ge 11 ] ; then
+ # For G1
+ JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ else
+ JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+ fi
else
JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"
fi
diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
index 53de822c..33e5e1cb 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
@@ -558,6 +558,11 @@ public class CanalController {
}
ZkClientx.clearClients();
+
+ // 需要释放 CanalServerWithEmbedded 否则主线程退出后,进程无法自动完整退出...
+ if (embededCanalServer != null) {
+ embededCanalServer.stop();
+ }
}
private void initCid(String path) {
diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties
index 82992bcc..e6bdc4cc 100644
--- a/deployer/src/main/resources/canal.properties
+++ b/deployer/src/main/resources/canal.properties
@@ -148,8 +148,13 @@ kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0
kafka.kerberos.enable = false
-kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
-kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
+kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
+kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf
+
+# sasl demo
+# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
+# kafka.sasl.mechanism = SCRAM-SHA-512
+# kafka.security.protocol = SASL_PLAINTEXT
##################################################
######### RocketMQ #############
@@ -179,4 +184,4 @@ rabbitmq.deliveryMode =
##################################################
pulsarmq.serverUrl =
pulsarmq.roleToken =
-pulsarmq.topicTenantPrefix =
\ No newline at end of file
+pulsarmq.topicTenantPrefix =
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java
index 5926a4fa..0eaea9cd 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java
@@ -181,9 +181,7 @@ public class UUIDSet {
int j;
for (j = i + 1; j < len; j++) {
if (intervals.get(i).stop >= intervals.get(j).start) {
- if (intervals.get(i).stop < intervals.get(j).stop) {
- intervals.get(i).stop = intervals.get(j).stop;
- }
+ intervals.get(i).stop = intervals.get(j).stop;
} else {
break;
}
diff --git a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java
index bf29f7d9..1c8ba0e9 100644
--- a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java
+++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlGTIDSetTest.java
@@ -42,10 +42,9 @@ public class MysqlGTIDSetTest {
MysqlGTIDSet mysqlGTIDSet1 = MysqlGTIDSet.parse(gtid1);
String gtid2 = "726757ad-4455-11e8-ae04-0242ac110002:1-20304074";
- MysqlGTIDSet mysqlGTIDSet2 = MysqlGTIDSet.parse(gtid2);
mysqlGTIDSet1.update(gtid2);
- assertEquals("726757ad-4455-11e8-ae04-0242ac110002:1-25536412", mysqlGTIDSet1.toString());
+ assertEquals("726757ad-4455-11e8-ae04-0242ac110002:1-20304074", mysqlGTIDSet1.toString());
}
@Test
diff --git a/example/src/main/bin/startup.sh b/example/src/main/bin/startup.sh
index 1fd8bf69..e83d7440 100644
--- a/example/src/main/bin/startup.sh
+++ b/example/src/main/bin/startup.sh
@@ -71,9 +71,12 @@ else
fi
if [ -n "$str" ]; then
- # JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
- # For G1
- JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ if [ $JavaVersion -ge 11 ] ; then
+ # For G1
+ JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
+ else
+ JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS"
+ fi
else
JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"
fi
diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java
index 48b51b25..cb20762a 100644
--- a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java
+++ b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java
@@ -109,11 +109,11 @@ public class AviaterRegexFilter implements CanalEventFilter {
private List completionPattern(List patterns) {
List result = new ArrayList<>();
for (String pattern : patterns) {
- StringBuffer stringBuffer = new StringBuffer();
- stringBuffer.append("^");
- stringBuffer.append(pattern);
- stringBuffer.append("$");
- result.add(stringBuffer.toString());
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("^");
+ stringBuilder.append(pattern);
+ stringBuilder.append("$");
+ result.add(stringBuilder.toString());
}
return result;
}
diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java
index 2f509569..f8fb4355 100644
--- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java
+++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/plain/PlainCanalConfigClient.java
@@ -16,6 +16,7 @@ import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.common.CanalException;
import com.alibaba.otter.canal.common.CanalLifeCycle;
import com.alibaba.otter.canal.protocol.SecurityUtil;
+import com.google.common.net.UrlEscapers;
/**
* 远程配置获取
@@ -73,7 +74,7 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
md5 = "";
}
String url = configURL + "/api/v1/config/server_polling?ip=" + localIp + "&port=" + adminPort + "&md5=" + md5
- + "®ister=" + (autoRegister ? 1 : 0) + "&cluster=" + autoCluster + "&name=" + name;
+ + "®ister=" + (autoRegister ? 1 : 0) + "&cluster=" + StringUtils.stripToEmpty(autoCluster) + "&name=" + StringUtils.stripToEmpty(name);
return queryConfig(url);
}
@@ -84,7 +85,7 @@ public class PlainCanalConfigClient extends AbstractCanalLifeCycle implements Ca
if (StringUtils.isEmpty(md5)) {
md5 = "";
}
- String url = configURL + "/api/v1/config/instance_polling/" + destination + "?md5=" + md5;
+ String url = configURL + "/api/v1/config/instance_polling/" + UrlEscapers.urlPathSegmentEscaper().escape(destination) + "?md5=" + md5;
return queryConfig(url);
}
diff --git a/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java b/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java
index 53d92701..052e923c 100644
--- a/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java
+++ b/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java
@@ -171,7 +171,7 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
return null;
}
- String json = FileUtils.readFileToString(dataFile, charset.name());
+ String json = FileUtils.readFileToString(dataFile, charset);
return JsonUtils.unmarshalFromString(json, FileMetaInstanceData.class);
} catch (IOException e) {
throw new CanalMetaManagerException(e);
@@ -209,7 +209,11 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
data.setClientDatas(clientDatas);
}
-
+ //fixed issue https://github.com/alibaba/canal/issues/4312
+ //客户端数据为空时不覆盖文件内容 (适合单客户端)
+ if(data.getClientDatas().isEmpty()){
+ return;
+ }
String json = JsonUtils.marshalToString(data);
try {
FileUtils.writeStringToFile(dataFile, json);
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 8636acfd..00f3d68e 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
@@ -12,6 +12,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import com.taobao.tddl.dbsync.binlog.event.*;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
@@ -40,23 +41,7 @@ import com.alibaba.otter.canal.protocol.CanalEntry.Type;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.google.protobuf.ByteString;
import com.taobao.tddl.dbsync.binlog.LogEvent;
-import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.HeartbeatLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.LogHeader;
-import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.RandLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer;
-import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
-import com.taobao.tddl.dbsync.binlog.event.UnknownLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidListLogEvent;
import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidLogEvent;
@@ -147,6 +132,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return parseGTIDLogEvent((GtidLogEvent) logEvent);
case LogEvent.HEARTBEAT_LOG_EVENT:
return parseHeartbeatLogEvent((HeartbeatLogEvent) logEvent);
+ case LogEvent.HEARTBEAT_LOG_EVENT_V2:
+ return parseHeartbeatV2LogEvent((HeartbeatV2LogEvent) logEvent);
case LogEvent.GTID_EVENT:
case LogEvent.GTID_LIST_EVENT:
return parseMariaGTIDLogEvent(logEvent);
@@ -173,13 +160,22 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return entryBuilder.build();
}
+ private Entry parseHeartbeatV2LogEvent(HeartbeatV2LogEvent logEvent) {
+ Header.Builder headerBuilder = Header.newBuilder();
+ headerBuilder.setEventType(EventType.MHEARTBEAT);
+ Entry.Builder entryBuilder = Entry.newBuilder();
+ entryBuilder.setHeader(headerBuilder.build());
+ entryBuilder.setEntryType(EntryType.HEARTBEAT);
+ return entryBuilder.build();
+ }
+
private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
LogHeader logHeader = logEvent.getHeader();
Pair.Builder builder = Pair.newBuilder();
builder.setKey("gtid");
builder.setValue(logEvent.getGtidStr());
- if (logEvent.getLastCommitted() != null) {
+ if (logEvent.getLastCommitted() != -1) {
builder.setKey("lastCommitted");
builder.setValue(String.valueOf(logEvent.getLastCommitted()));
builder.setKey("sequenceNumber");
@@ -396,7 +392,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
// mysql5.6支持,需要设置binlog-rows-query-log-events=1,可详细打印原始DML语句
String queryString = null;
try {
- queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset.name());
+ queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
String tableName = null;
if (useDruidDdlFilter) {
List results = DruidDdlParser.parse(queryString, null);
@@ -418,7 +414,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
// mariaDb支持,需要设置binlog_annotate_row_events=true,可详细打印原始DML语句
String queryString = null;
try {
- queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset.name());
+ queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
return buildQueryEntry(queryString, event.getHeader());
} catch (UnsupportedEncodingException e) {
throw new CanalParseException(e);
@@ -512,10 +508,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
public void parseTableMapEvent(TableMapLogEvent event) {
try {
- String charsetDbName = new String(event.getDbName().getBytes(ISO_8859_1), charset.name());
+ String charsetDbName = new String(event.getDbName().getBytes(ISO_8859_1), charset);
event.setDbname(charsetDbName);
- String charsetTbName = new String(event.getTableName().getBytes(ISO_8859_1), charset.name());
+ String charsetTbName = new String(event.getTableName().getBytes(ISO_8859_1), charset);
event.setTblname(charsetTbName);
} catch (UnsupportedEncodingException e) {
throw new CanalParseException(e);
@@ -554,7 +550,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
rowChangeBuider.setIsDdl(false);
rowChangeBuider.setEventType(eventType);
- RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+ RowsLogBuffer buffer = event.getRowsBuf(charset);
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java
index 9a7a5bb5..889b359a 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java
@@ -100,7 +100,7 @@ public class DdlResult {
@Override
public String toString() {
DdlResult ddlResult = this;
- StringBuffer sb = new StringBuffer();
+ StringBuilder sb = new StringBuilder();
do {
sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
ddlResult.schemaName,
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 b4c7ea2f..0a2eb20e 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
@@ -209,6 +209,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
for (String schema : schemas) {
// filter views
packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
+ columnSize = packet.getFieldDescriptors().size();
int tableNameColumnIndex = 0; // default index is 0
List tables = new ArrayList<>();
for (int line = 0; line < packet.getFieldValues().size() / columnSize; line++) {
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 55aa702d..1d27f370 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
@@ -80,7 +80,8 @@ public class MemoryTableMeta implements TableMetaTSDB {
&& !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create user")
&& !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "alter user")
&& !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "drop user")
- && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create database")) {
+ && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "create database")
+ && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "alter schema")) {
repository.console(ddl);
}
} catch (Throwable e) {
@@ -145,7 +146,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
public Map snapshot() {
Map schemaDdls = new HashMap<>();
for (Schema schema : repository.getSchemas()) {
- StringBuffer data = new StringBuffer(4 * 1024);
+ StringBuilder data = new StringBuilder(4 * 1024);
for (String table : schema.showTables()) {
SchemaObject schemaObject = schema.findTable(table);
schemaObject.getStatement().output(data);
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
index 3a0fedd6..4a0ab4a8 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
@@ -182,7 +182,7 @@ public class FileMixedLogPositionManager extends AbstractLogPositionManager {
return null;
}
- String json = FileUtils.readFileToString(dataFile, charset.name());
+ String json = FileUtils.readFileToString(dataFile, charset);
return JsonUtils.unmarshalFromString(json, LogPosition.class);
} catch (IOException e) {
throw new CanalMetaManagerException(e);
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
index 8207f4b7..1fe0010b 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
@@ -302,7 +302,7 @@ public class DirectLogFetcherTest {
event.getHeader().getLogPos() - event.getHeader().getEventLen(),
event.getTable().getDbName(),
event.getTable().getTableName()));
- RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+ RowsLogBuffer buffer = event.getRowsBuf(charset);
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java
index 58a9a9a4..c8cef505 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java
@@ -134,7 +134,7 @@ public class MysqlBinlogParsePerformanceTest {
public static void parseRowsEvent(RowsLogEvent event, AtomicLong sum) {
try {
- RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+ RowsLogBuffer buffer = event.getRowsBuf(charset);
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {
diff --git a/pom.xml b/pom.xml
index 1597f67d..0416fcbc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -58,7 +58,7 @@