chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #892 from rewerma/master

Browse Source 
kafka client keep alive间隔时间
This commit is contained in:
agapple
2018-08-28 17:47:31 +08:00
committed by GitHub
parent 9434ed513e bcf30a9a05
commit fd5cfe23db
3 changed files with 7 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java
View File

@@ -143,9 +143,13 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
}); });
// 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
long currentTS = System.currentTimeMillis();
while (executing.get()) { while (executing.get()) {
connector.ack(); // 大于1分钟未消费完ack一次keep alive
Thread.sleep(500); if (System.currentTimeMillis() - currentTS > 60000) {
connector.ack();
currentTS = System.currentTimeMillis();
}
} }
} else { } else {
connector.ack(); connector.ack();

2
deployer/src/main/resources/kafka.yml
View File

@@ -3,8 +3,6 @@ retries: 0
batchSize: 16384 batchSize: 16384
lingerMs: 1 lingerMs: 1
bufferMemory: 33554432 bufferMemory: 33554432
# canal的批次大小，单位 k，量大建议改为1M
canalBatchSize: 50
filterTransactionEntry: true filterTransactionEntry: true
canalDestinations: canalDestinations:

2
server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java
View File

@@ -19,7 +19,7 @@ public class KafkaProperties {
private int lingerMs = 1; private int lingerMs = 1;
private long bufferMemory = 33554432L; private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true; private boolean filterTransactionEntry = true;
private int canalBatchSize = 5; private int canalBatchSize = 50;
private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>(); private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>();