consumer group: fix time-sleep mechanism

aliyun · Dec 30, 2018 · b012c1d · b012c1d
1 parent e23d0f9
commit b012c1d
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 3 deletions.
diff --git a/aliyun/log/consumer/heart_beat.py b/aliyun/log/consumer/heart_beat.py
@@ -20,18 +20,17 @@ def __init__(self, log_client, heartbeat_interval):
 
     def run(self):
         logger.info('heart beat start')
-        last_heatbeat_time = 0
         while not self.shut_down_flag:
             try:
                 response_shards = []
+                last_heatbeat_time = time.time()
                 self.log_client.heartbeat(self.mheart_shards, response_shards)
                 logger.info('heart beat result: {} get: {}'.format(self.mheart_shards, response_shards))
                 self.mheld_shards = response_shards
                 self.mheart_shards = self.mheld_shards[:]
 
                 # default sleep for 2s from "LogHubConfig"
                 time_to_sleep = self.heartbeat_interval - (time.time() - last_heatbeat_time)
-                last_heatbeat_time = time.time()
                 if time_to_sleep > 0 and not self.shut_down_flag:
                     time.sleep(time_to_sleep)
             except Exception as e:

diff --git a/aliyun/log/consumer/worker.py b/aliyun/log/consumer/worker.py
@@ -41,6 +41,7 @@ def run(self):
         while not self.shut_down_flag:
             held_shards = self.heart_beat.get_held_shards()
 
+            last_fetch_time = time.time()
             for shard in held_shards:
                 if self.shut_down_flag:
                     break
@@ -52,7 +53,6 @@ def run(self):
 
             # default sleep for 2s from "LogHubConfig"
             time_to_sleep = self.option.data_fetch_interval - (time.time() - last_fetch_time)
-            last_fetch_time = time.time()
             if time_to_sleep > 0 and not self.shut_down_flag:
                 time.sleep(time_to_sleep)