diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java index 5bccb51a..aecaceb2 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -143,9 +143,13 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { }); // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 + long currentTS = System.currentTimeMillis(); while (executing.get()) { - connector.ack(); - Thread.sleep(500); + // 大于1分钟未消费完ack一次keep alive + if (System.currentTimeMillis() - currentTS > 60000) { + connector.ack(); + currentTS = System.currentTimeMillis(); + } } } else { connector.ack(); diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 93d4d557..8abf4934 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -3,8 +3,6 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 -# canal的批次大小,单位 k,量大建议改为1M -canalBatchSize: 50 filterTransactionEntry: true canalDestinations: diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 140ce965..9194c0ab 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -19,7 +19,7 @@ public class KafkaProperties { private int lingerMs = 1; private long bufferMemory = 33554432L; private boolean filterTransactionEntry = true; - private int canalBatchSize = 5; + private int canalBatchSize = 50; private List canalDestinations = new ArrayList();