Merge pull request #542 from trauter/rhmaster2

Fix some cases of hanging rebalance
robinhood · Oct 9, 2020 · 01b4c0a · 01b4c0a
2 parents b5e159f + e7b02e8
commit 01b4c0a
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py
@@ -743,6 +743,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
         ])