diff --git a/client-adapter/common/pom.xml b/client-adapter/common/pom.xml index e62fc386..8a6d75fa 100644 --- a/client-adapter/common/pom.xml +++ b/client-adapter/common/pom.xml @@ -3,7 +3,7 @@ canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter @@ -14,7 +14,7 @@ com.alibaba.otter canal.protocol - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT joda-time diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index 672d83ec..3574b678 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -30,6 +30,10 @@ public class CanalClientConfig { private Long timeout; // 模式 tcp kafka rocketMQ private String mode = "tcp"; + // aliyun ak/sk + private String accessKey; + private String secretKey; + // canal adapters 配置 private List canalAdapters; @@ -105,6 +109,22 @@ public class CanalClientConfig { this.mode = mode; } + public String getAccessKey() { + return accessKey; + } + + public void setAccessKey(String accessKey) { + this.accessKey = accessKey; + } + + public String getSecretKey() { + return secretKey; + } + + public void setSecretKey(String secretKey) { + this.secretKey = secretKey; + } + public List getCanalAdapters() { return canalAdapters; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java index 4e735c02..41d1864e 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java @@ -27,8 +27,7 @@ import org.slf4j.LoggerFactory; */ public class ExtensionLoader { - private static final Logger logger = LoggerFactory - .getLogger(ExtensionLoader.class); + private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class); private static final String SERVICES_DIRECTORY = "META-INF/services/"; @@ -36,8 +35,7 @@ public class ExtensionLoader { private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; - private static final Pattern NAME_SEPARATOR = Pattern - .compile("\\s*[,]+\\s*"); + private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*"); private static final ConcurrentMap, ExtensionLoader> EXTENSION_LOADERS = new ConcurrentHashMap<>(); @@ -173,8 +171,7 @@ public class ExtensionLoader { return instance; } catch (Throwable t) { throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type - + ") could not be instantiated: " + t.getMessage(), - t); + + ") could not be instantiated: " + t.getMessage(), t); } } @@ -194,8 +191,7 @@ public class ExtensionLoader { return instance; } catch (Throwable t) { throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type - + ") could not be instantiated: " + t.getMessage(), - t); + + ") could not be instantiated: " + t.getMessage(), t); } } @@ -334,10 +330,12 @@ public class ExtensionLoader { // Class.forName(line, true, // classLoader); if (!type.isAssignableFrom(clazz)) { - throw new IllegalStateException( - "Error when load extension class(interface: " + type - + ", class line: " + clazz.getName() - + "), class " + clazz.getName() + throw new IllegalStateException("Error when load extension class(interface: " + + type + + ", class line: " + + clazz.getName() + + "), class " + + clazz.getName() + "is not subtype of interface."); } else { try { @@ -355,9 +353,9 @@ public class ExtensionLoader { extensionClasses.put(n, clazz); } else if (c != clazz) { cachedNames.remove(clazz); - throw new IllegalStateException( - "Duplicate extension " + type.getName() + " name " - + n + " on " + throw new IllegalStateException("Duplicate extension " + + type.getName() + + " name " + n + " on " + c.getName() + " and " + clazz.getName()); } @@ -367,9 +365,12 @@ public class ExtensionLoader { } } } catch (Throwable t) { - IllegalStateException e = new IllegalStateException( - "Failed to load extension class(interface: " + type + ", class line: " - + line + ") in " + url + IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: " + + type + + ", class line: " + + line + + ") in " + + url + ", cause: " + t.getMessage(), t); @@ -384,18 +385,17 @@ public class ExtensionLoader { } } catch (Throwable t) { logger.error("Exception when load extension class(interface: " + type + ", class file: " + url - + ") in " + url, - t); + + ") in " + url, t); } } // end of while urls } } catch (Throwable t) { - logger.error( - "Exception when load extension class(interface: " + type + ", description file: " + fileName + ").", - t); + logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName + + ").", t); } } + @SuppressWarnings("unused") private static ClassLoader findClassLoader() { return ExtensionLoader.class.getClassLoader(); } diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml index 2d6c3def..35711014 100644 --- a/client-adapter/elasticsearch/pom.xml +++ b/client-adapter/elasticsearch/pom.xml @@ -1,11 +1,9 @@ - + canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter @@ -81,9 +79,9 @@ - + - + diff --git a/client-adapter/hbase/pom.xml b/client-adapter/hbase/pom.xml index 10d08681..29ee8b33 100644 --- a/client-adapter/hbase/pom.xml +++ b/client-adapter/hbase/pom.xml @@ -3,7 +3,7 @@ canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter @@ -72,9 +72,9 @@ - + - + diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index e1f557ce..32c70028 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -1,10 +1,9 @@ - + canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter @@ -33,7 +32,7 @@ com.alibaba.otter canal.client - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT org.yaml diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java index 83f4b350..3d2ef585 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/CanalAdapterApplication.java @@ -3,7 +3,6 @@ package com.alibaba.otter.canal.adapter.launcher; import org.springframework.boot.Banner; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.builder.SpringApplicationBuilder; /** * 启动入口 diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java index aa6e55bc..c0e5d394 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/common/SyncSwitch.java @@ -64,11 +64,12 @@ public class SyncSwitch { } } + @SuppressWarnings("resource") private synchronized void startListen(String destination, BooleanMutex mutex) { try { String path = SYN_SWITCH_ZK_NODE + destination; CuratorFramework curator = curatorClient.getCurator(); - final NodeCache nodeCache = new NodeCache(curator, path); + NodeCache nodeCache = new NodeCache(curator, path); nodeCache.start(); nodeCache.getListenable().addListener(() -> initMutex(curator, destination, mutex)); } catch (Exception e) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java index 2b9bb243..3ad0d5f7 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/config/AdapterCanalConfig.java @@ -47,6 +47,7 @@ public class AdapterCanalConfig extends CanalClientConfig { return srcDataSources; } + @SuppressWarnings("resource") public void setSrcDataSources(Map srcDataSources) { this.srcDataSources = srcDataSources; diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java index 631154c2..050bbf65 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterKafkaWorker.java @@ -44,8 +44,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { while (!running) ; ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetries() == null - || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); + int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 162cdc11..80d5371a 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -122,6 +122,8 @@ public class CanalAdapterLoader { canalAdapter.getInstance(), group.getGroupId(), canalOuterAdapterGroups, + canalClientConfig.getAccessKey(), + canalClientConfig.getSecretKey(), canalClientConfig.getFlatMessage()); canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); rocketMQWorker.start(); diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java index 7f2e3b56..2e737353 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java @@ -3,16 +3,12 @@ package com.alibaba.otter.canal.adapter.launcher.loader; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import org.apache.kafka.common.errors.WakeupException; import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; -import com.alibaba.otter.canal.protocol.FlatMessage; -import com.alibaba.otter.canal.protocol.Message; /** * rocketmq对应的client适配器工作线程 @@ -26,13 +22,14 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { private boolean flatMessage; public CanalAdapterRocketMQWorker(CanalClientConfig canalClientConfig, String nameServers, String topic, - String groupId, List> canalOuterAdapters, boolean flatMessage){ + String groupId, List> canalOuterAdapters, String accessKey, + String secretKey, boolean flatMessage){ super(canalOuterAdapters); this.canalClientConfig = canalClientConfig; this.topic = topic; this.flatMessage = flatMessage; this.canalDestination = topic; - this.connector = new RocketMQCanalConnector(nameServers, topic, groupId, flatMessage); + this.connector = new RocketMQCanalConnector(nameServers, topic, groupId, accessKey, secretKey, flatMessage); logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); } @@ -42,8 +39,7 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker { ; ExecutorService workerExecutor = Executors.newSingleThreadExecutor(); - int retry = canalClientConfig.getRetries() == null - || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); + int retry = canalClientConfig.getRetries() == null || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 while (running) { diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 5495a4f8..098bc1f3 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -21,6 +21,8 @@ canal.conf: syncBatchSize: 1000 retries: 0 timeout: + accessKey: + secretKey: mode: tcp # kafka rocketMQ # srcDataSources: # defaultDS: diff --git a/client-adapter/logger/pom.xml b/client-adapter/logger/pom.xml index bca319f1..11c8d53e 100644 --- a/client-adapter/logger/pom.xml +++ b/client-adapter/logger/pom.xml @@ -3,7 +3,7 @@ canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml index 2f12209c..dd216b25 100644 --- a/client-adapter/pom.xml +++ b/client-adapter/pom.xml @@ -3,9 +3,14 @@ 4.0.0 com.alibaba.otter canal.client-adapter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT pom canal client adapter module for otter ${project.version} + + com.alibaba.otter + canal + 1.1.3-SNAPSHOT + UTF-8 @@ -25,9 +30,42 @@ rdb + + + Apache License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0 + + + + + git@github.com:alibaba/canal.git + scm:git:git@github.com:alibaba/canal.git + scm:git:git@github.com:alibaba/canal.git + + central + http://repo1.maven.org/maven2 + + true + + + false + + + + java.net + http://download.java.net/maven/2/ + + true + + + false + + + + aliyun http://maven.aliyun.com/nexus/content/groups/public/ true @@ -36,6 +74,28 @@ false + + sonatype + sonatype + https://oss.sonatype.org/content/repositories/snapshots + + false + + + true + + + + sonatype-release + sonatype-release + https://oss.sonatype.org/service/local/repositories/releases/content + + false + + + true + + @@ -52,4 +112,17 @@ - + + + + sonatype-nexus-snapshots + Sonatype Nexus Snapshots + https://oss.sonatype.org/content/repositories/snapshots/ + + + sonatype-nexus-staging + Nexus Release Repository + https://oss.sonatype.org/service/local/staging/deploy/maven2/ + + + \ No newline at end of file diff --git a/client-adapter/rdb/pom.xml b/client-adapter/rdb/pom.xml index 36881768..458e8ab2 100644 --- a/client-adapter/rdb/pom.xml +++ b/client-adapter/rdb/pom.xml @@ -1,11 +1,9 @@ - + canal.client-adapter com.alibaba.otter - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT 4.0.0 com.alibaba.otter @@ -84,9 +82,9 @@ - + - + diff --git a/client/pom.xml b/client/pom.xml index 28ca8d27..3d029080 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT ../pom.xml com.alibaba.otter diff --git a/common/pom.xml b/common/pom.xml index 9a3c3b22..9d6777b3 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -3,7 +3,7 @@ com.alibaba.otter canal - 1.1.2-SNAPSHOT + 1.1.3-SNAPSHOT ../pom.xml canal.common diff --git a/common/src/main/java/com/alibaba/otter/canal/common/utils/BooleanMutex.java b/common/src/main/java/com/alibaba/otter/canal/common/utils/BooleanMutex.java index 660181ce..3a2ee258 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/utils/BooleanMutex.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/utils/BooleanMutex.java @@ -5,7 +5,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.AbstractQueuedSynchronizer; /** - * 实现一个互斥实现，基于Cocurrent中的AbstractQueuedSynchronizer实现了自己的sync
+ * 实现一个互斥实现，基于Cocurrent中的AQS实现了自己的sync
* 应用场景：系统初始化/授权控制，没权限时阻塞等待。有权限时所有线程都可以快速通过 * * 
diff --git a/dbsync/pom.xml b/dbsync/pom.xml
index 9aa0fa68..4b027523 100644
--- a/dbsync/pom.xml
+++ b/dbsync/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/deployer/pom.xml b/deployer/pom.xml
index 25d0c6e3..d7de9395 100644
--- a/deployer/pom.xml
+++ b/deployer/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
index d21348bf..ba39bde0 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
@@ -41,6 +41,7 @@ public class CanalConstants {
     public static final String CANAL_MQ_RETRIES                  = ROOT + "." + "mq.retries";
     public static final String CANAL_MQ_BATCHSIZE                = ROOT + "." + "mq.batchSize";
     public static final String CANAL_MQ_LINGERMS                 = ROOT + "." + "mq.lingerMs";
+    public static final String CANAL_MQ_MAXREQUESTSIZE           = ROOT + "." + "mq.maxRequestSize";
     public static final String CANAL_MQ_BUFFERMEMORY             = ROOT + "." + "mq.bufferMemory";
     public static final String CANAL_MQ_CANALBATCHSIZE           = ROOT + "." + "mq.canalBatchSize";
     public static final String CANAL_MQ_CANALGETTIMEOUT          = ROOT + "." + "mq.canalGetTimeout";
diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
index 24d0ec5b..aa7f53b0 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java
@@ -101,6 +101,10 @@ public class CanalLauncher {
         if (!StringUtils.isEmpty(lingerMs)) {
             mqProperties.setLingerMs(Integer.valueOf(lingerMs));
         }
+        String maxRequestSize = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_MAXREQUESTSIZE);
+        if (!StringUtils.isEmpty(maxRequestSize)) {
+            mqProperties.setMaxRequestSize(Integer.valueOf(maxRequestSize));
+        }
         String bufferMemory = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_BUFFERMEMORY);
         if (!StringUtils.isEmpty(bufferMemory)) {
             mqProperties.setBufferMemory(Long.valueOf(bufferMemory));
diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties
index 78343b43..5d7b9d06 100644
--- a/deployer/src/main/resources/canal.properties
+++ b/deployer/src/main/resources/canal.properties
@@ -104,6 +104,7 @@ canal.instance.global.spring.xml = classpath:spring/file-instance.xml
 canal.mq.servers = 127.0.0.1:6667
 canal.mq.retries = 0
 canal.mq.batchSize = 16384
+canal.mq.maxRequestSize = 1048576
 canal.mq.lingerMs = 1
 canal.mq.bufferMemory = 33554432
 canal.mq.canalBatchSize = 50
diff --git a/driver/pom.xml b/driver/pom.xml
index b7db03fe..05baa5ac 100644
--- a/driver/pom.xml
+++ b/driver/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/example/pom.xml b/example/pom.xml
index ae040df6..819b4b12 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/filter/pom.xml b/filter/pom.xml
index e9a2f60c..6fc6ba88 100644
--- a/filter/pom.xml
+++ b/filter/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/instance/core/pom.xml b/instance/core/pom.xml
index c96d8ce5..a7346fc8 100644
--- a/instance/core/pom.xml
+++ b/instance/core/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../../pom.xml
 	
 	canal.instance.core
diff --git a/instance/manager/pom.xml b/instance/manager/pom.xml
index 0710dbd1..ef5c987e 100644
--- a/instance/manager/pom.xml
+++ b/instance/manager/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../../pom.xml
 	
 	com.alibaba.otter
diff --git a/instance/pom.xml b/instance/pom.xml
index 3ccf7ebd..2a4da45c 100644
--- a/instance/pom.xml
+++ b/instance/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/instance/spring/pom.xml b/instance/spring/pom.xml
index 3d5526d4..bea5d37d 100644
--- a/instance/spring/pom.xml
+++ b/instance/spring/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../../pom.xml
 	
 	com.alibaba.otter
diff --git a/meta/pom.xml b/meta/pom.xml
index 8683e8e7..002f6231 100644
--- a/meta/pom.xml
+++ b/meta/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/parse/pom.xml b/parse/pom.xml
index d610838d..985d5de9 100644
--- a/parse/pom.xml
+++ b/parse/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	canal.parse
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
index 5e94e40b..af792ec8 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
@@ -3,9 +3,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql;
 import java.nio.charset.Charset;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.alibaba.otter.canal.filter.CanalEventFilter;
 import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
 import com.alibaba.otter.canal.parse.CanalEventParser;
@@ -21,7 +18,6 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition;
 
 public abstract class AbstractMysqlEventParser extends AbstractEventParser {
 
-    protected final Logger         logger                    = LoggerFactory.getLogger(this.getClass());
     protected static final long    BINLOG_START_OFFEST       = 4L;
 
     protected TableMetaTSDBFactory tableMetaTSDBFactory      = new DefaultTableMetaTSDBFactory();
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
index 0528d19c..35e98f73 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
@@ -135,6 +135,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
     protected void afterDump(ErosaConnection connection) {
         super.afterDump(connection);
 
+        if (connection == null) {
+            throw new CanalParseException("illegal connection is null");
+        }
+
         if (!(connection instanceof MysqlConnection)) {
             throw new CanalParseException("Unsupported connection type : " + connection.getClass().getSimpleName());
         }
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 e9bdc2ad..033dcae5 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
@@ -76,7 +76,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
     private int                             snapshotInterval = 24;
     private int                             snapshotExpire   = 360;
     private ScheduledFuture              scheduleSnapshotFuture;
-    
+
     public DatabaseTableMeta(){
 
     }
@@ -207,8 +207,8 @@ public class DatabaseTableMeta implements TableMetaTSDB {
                 packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
                 List tables = new ArrayList();
                 for (String table : packet.getFieldValues()) {
-                    if("BASE TABLE".equalsIgnoreCase(table)){
-                       continue; 
+                    if ("BASE TABLE".equalsIgnoreCase(table)) {
+                        continue;
                     }
                     String fullName = schema + "." + table;
                     if (blackFilter == null || !blackFilter.filter(fullName)) {
@@ -310,11 +310,19 @@ public class DatabaseTableMeta implements TableMetaTSDB {
         boolean compareAll = true;
         for (Schema schema : tmpMemoryTableMeta.getRepository().getSchemas()) {
             for (String table : schema.showTables()) {
-                if (!compareTableMetaDbAndMemory(connection, tmpMemoryTableMeta, schema.getName(), table)) {
-                    compareAll = false;
+                String fullName = schema + "." + table;
+                if (blackFilter == null || !blackFilter.filter(fullName)) {
+                    if (filter == null || filter.filter(fullName)) {
+                        // issue : https://github.com/alibaba/canal/issues/1168
+                        // 在生成snapshot时重新过滤一遍
+                        if (!compareTableMetaDbAndMemory(connection, tmpMemoryTableMeta, schema.getName(), table)) {
+                            compareAll = false;
+                        }
+                    }
                 }
             }
         }
+
         if (compareAll) {
             Map content = new HashMap();
             content.put("destination", destination);
@@ -513,7 +521,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
             String sign = sourceField.isUnsigned() ? "unsigned" : "signed";
             String sourceColumnType = StringUtils.removeEndIgnoreCase(sourceField.getColumnType(), sign).trim();
             String targetColumnType = StringUtils.removeEndIgnoreCase(targetField.getColumnType(), sign).trim();
-            
+
             boolean columnTypeCompare = false;
             columnTypeCompare |= StringUtils.containsIgnoreCase(sourceColumnType, targetColumnType);
             columnTypeCompare |= StringUtils.containsIgnoreCase(targetColumnType, sourceColumnType);
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 719e86dd..089163d3 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
@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.parse;
 
+import static com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.MASTER_HEARTBEAT_PERIOD_SECONDS;
+
 import java.io.IOException;
 import java.io.Serializable;
 import java.io.UnsupportedEncodingException;
@@ -7,6 +9,7 @@ import java.net.InetSocketAddress;
 import java.nio.charset.Charset;
 import java.util.BitSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang.StringUtils;
 import org.junit.Assert;
@@ -213,6 +216,13 @@ public class DirectLogFetcherTest {
         } catch (Exception e) {
             logger.warn("update mariadb_slave_capability failed", e);
         }
+
+        try {
+            long period = TimeUnit.SECONDS.toNanos(MASTER_HEARTBEAT_PERIOD_SECONDS);
+            update("SET @master_heartbeat_period=" + period, connector);
+        } catch (Exception e) {
+            logger.warn("update master_heartbeat_period failed", e);
+        }
     }
 
     private void loadBinlogChecksum(MysqlConnector connector) {
diff --git a/pom.xml b/pom.xml
index 92d8a530..e5ba86c9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -4,7 +4,7 @@
     canal
     pom
     canal module for otter ${project.version}
-    1.1.2-SNAPSHOT
+    1.1.3-SNAPSHOT
     https://github.com/alibaba/canal
     
         org.sonatype.oss
@@ -381,7 +381,6 @@
                 
             
             
-            
         
         src/main/java
         src/test/java
diff --git a/prometheus/pom.xml b/prometheus/pom.xml
index 0e0b2217..f6357c20 100644
--- a/prometheus/pom.xml
+++ b/prometheus/pom.xml
@@ -3,13 +3,13 @@
     
         canal
         com.alibaba.otter
-        1.1.2-SNAPSHOT
+        1.1.3-SNAPSHOT
     
     4.0.0
 
     com.alibaba.otter
     canal.prometheus
-    1.1.2-SNAPSHOT
+    1.1.3-SNAPSHOT
     canal prometheus module for otter ${project.version}
     
         
diff --git a/protocol/pom.xml b/protocol/pom.xml
index f1cdfa2f..1b776b7e 100644
--- a/protocol/pom.xml
+++ b/protocol/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/server/pom.xml b/server/pom.xml
index fd1225f4..709a14e0 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	canal.server
diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
index cf72a2dd..a337ccc6 100644
--- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
+++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
@@ -14,6 +14,7 @@ public class MQProperties {
     private int     retries                = 0;
     private int     batchSize              = 16384;
     private int     lingerMs               = 1;
+    private int     maxRequestSize         = 1048576;
     private long    bufferMemory           = 33554432L;
     private boolean filterTransactionEntry = true;
     private String  producerGroup          = "Canal-Producer";
@@ -185,4 +186,11 @@ public class MQProperties {
     public void setAliyunSecretKey(String aliyunSecretKey) {
         this.aliyunSecretKey = aliyunSecretKey;
     }
+    public int getMaxRequestSize() {
+        return maxRequestSize;
+    }
+
+    public void setMaxRequestSize(int maxRequestSize) {
+        this.maxRequestSize = maxRequestSize;
+    }
 }
diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
index bf97de8e..230e9607 100644
--- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
@@ -43,6 +43,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
         properties.put("retries", kafkaProperties.getRetries());
         properties.put("batch.size", kafkaProperties.getBatchSize());
         properties.put("linger.ms", kafkaProperties.getLingerMs());
+        properties.put("max.request.size", kafkaProperties.getMaxRequestSize());
         properties.put("buffer.memory", kafkaProperties.getBufferMemory());
         properties.put("key.serializer", StringSerializer.class.getName());
         if (!kafkaProperties.getFlatMessage()) {
diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
index ad4fc266..ade0adaf 100644
--- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
+++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
@@ -13,7 +13,6 @@ import org.slf4j.LoggerFactory;
 import com.alibaba.otter.canal.common.MQProperties;
 import com.alibaba.otter.canal.instance.core.CanalInstance;
 import com.alibaba.otter.canal.instance.core.CanalMQConfig;
-import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
 import com.alibaba.otter.canal.protocol.ClientIdentity;
 import com.alibaba.otter.canal.protocol.Message;
 import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
@@ -146,7 +145,7 @@ public class CanalMQStarter {
                     try {
                         int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size();
                         if (batchId != -1 && size != 0) {
-                            canalMQProducer.send(destination, message, new CanalKafkaProducer.Callback() {
+                            canalMQProducer.send(destination, message, new CanalMQProducer.Callback() {
 
                                 @Override
                                 public void commit() {
diff --git a/sink/pom.xml b/sink/pom.xml
index e4f9674d..5e706819 100644
--- a/sink/pom.xml
+++ b/sink/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/store/pom.xml b/store/pom.xml
index 531b39a4..4ff3d1e1 100644
--- a/store/pom.xml
+++ b/store/pom.xml
@@ -3,7 +3,7 @@
 	
 		com.alibaba.otter
 		canal
-		1.1.2-SNAPSHOT
+		1.1.3-SNAPSHOT
 		../pom.xml
 	
 	com.alibaba.otter
diff --git a/store/src/main/java/com/alibaba/otter/canal/store/model/Event.java b/store/src/main/java/com/alibaba/otter/canal/store/model/Event.java
index edb4dd24..db1618e1 100644
--- a/store/src/main/java/com/alibaba/otter/canal/store/model/Event.java
+++ b/store/src/main/java/com/alibaba/otter/canal/store/model/Event.java
@@ -71,8 +71,8 @@ public class Event implements Serializable {
             this.rawLength = rawEntry.size();
         } else {
             this.entry = entry;
-            // 按照3倍的event length预估
-            this.rawLength = entry.getHeader().getEventLength() * 3;
+            // 按照6倍的event length预估
+            this.rawLength = entry.getHeader().getEventLength() * 6;
         }
     }