From b5e05c8c77fd7091901500360e28f568b20ca89b Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Tue, 28 Aug 2018 13:51:23 +0800
Subject: [PATCH] =?UTF-8?q?kafka=20client=20keep=20alive=E9=97=B4=E9=9A=94?=
 =?UTF-8?q?=E6=97=B6=E9=97=B4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../client/adapter/loader/CanalAdapterKafkaWorker.java    | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

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();