#523: add commit.last_poll_timestamp(partition)

Author: aure-olli

CHANGES.rst

@@ -1,6 +1,12 @@
 CHANGES
 -------
 
+523.feature
+^^^^^^^^^^^
+
+Add `consumer.last_poll_timestamp(partition)` which gives the ms timestamp of the last update of `highwater` and `lso`.
+
+
 0.5.2 (2019-03-10)
 ^^^^^^^^^^^^^^^^^^
 

aiokafka/consumer/consumer.py

@@ -683,6 +683,25 @@ def last_stable_offset(self, partition):
         assignment = self._subscription.subscription.assignment
         return assignment.state_value(partition).lso
 
+    def last_poll_timestamp(self, partition):
+        """ Returns the timestamp of the last poll of this partition (in ms).
+        It is the last time `highwater` and `last_stable_offset` were
+        udpated. However it does not mean that new messages were received.
+
+        As with ``highwater()`` will not be available until some messages are
+        consumed.
+
+        Arguments:
+            partition (TopicPartition): partition to check
+
+        Returns:
+            int or None: timestamp if available
+        """
+        assert self._subscription.is_assigned(partition), \
+            'Partition is not assigned'
+        assignment = self._subscription.subscription.assignment
+        return assignment.state_value(partition).timestamp
+
     def seek(self, partition, offset):
         """ Manually specify the fetch offset for a TopicPartition.
 

aiokafka/consumer/fetcher.py

@@ -2,6 +2,7 @@
 import collections
 import logging
 import random
+import time
 from itertools import chain
 
 from kafka.protocol.offset import OffsetRequest
@@ -673,6 +674,7 @@ async def _proc_fetch_request(self, assignment, node_id, request):
             for partition, offset, _ in partitions:
                 fetch_offsets[TopicPartition(topic, partition)] = offset
 
+        now_ms = int(1000 * time.time())
         for topic, partitions in response.topics:
             for partition, error_code, highwater, *part_data in partitions:
                 tp = TopicPartition(topic, partition)
@@ -696,6 +698,7 @@ async def _proc_fetch_request(self, assignment, node_id, request):
                         lso = None
                     tp_state.highwater = highwater
                     tp_state.lso = lso
+                    tp_state.timestamp = now_ms
 
                     # part_data also contains lso, aborted_transactions.
                     # message_set is last

aiokafka/consumer/subscription_state.py

@@ -463,6 +463,7 @@ def __init__(self, assignment, *, loop):
 
         self.highwater = None  # Last fetched highwater mark
         self.lso = None  # Last fetched stable offset mark
+        self.timestamp = None  # timestamp of last poll
         self._position = None  # The current position of the topic
         self._position_fut = create_future(loop=loop)
 

docs/consumer.rst

@@ -444,8 +444,10 @@ catch up). For example::
         for partition in consumer.assignment():
             highwater = consumer.highwater(partition)
             position = await consumer.position(partition)
-            lag = highwater - position
-            if lag > LAG_THRESHOLD:
+            position_lag = highwater - position
+            timestamp = consumer.last_poll_timestamp(partition)
+            time_lag = time.time() * 1000 - timestamp
+            if position_lag > POSITION_THRESHOLD or time_lag > TIME_THRESHOLD:
                 partitions.append(partition)
 
 .. note:: This interface differs from `pause()`/`resume()` interface of 

tests/test_consumer.py

@@ -69,6 +69,7 @@ async def test_simple_consumer(self):
             # check unsupported version
             consumer = await self.consumer_factory(api_version="0.8")
 
+        now = time.time()
         await self.send_messages(0, list(range(0, 100)))
         await self.send_messages(1, list(range(100, 200)))
         # Start a consumer_factory
@@ -101,6 +102,10 @@ async def test_simple_consumer(self):
 
         h = consumer.highwater(p0)
         self.assertEqual(h, 100)
+        t = consumer.last_poll_timestamp(p0)
+        self.assertGreaterEqual(t, int(now * 1000))
+        now = time.time()
+        self.assertLessEqual(t, int(now * 1000))
 
         consumer.seek(p0, offset + 90)
         for i in range(10):