enable offset resetting to 0 (used for recovery)

robinhood · Apr 13, 2020 · e7b02e8 · e7b02e8
1 parent 38d61a7
commit e7b02e8
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py
@@ -741,6 +741,8 @@ async def _seek_wait(self,
             consumer.seek(tp, offset)
             if offset > 0:
                 self.consumer._read_offset[tp] = offset
+            elif tp in self.consumer._read_offset.keys():
+                del self.consumer._read_offset[tp]
         await asyncio.gather(*[
             consumer.position(tp) for tp in partitions
         ])