fix condition to delete old table keys (#251)

faust-streaming · Jan 6, 2022 · a0e9a31 · a0e9a31
1 parent 9805612
commit a0e9a31
Showing 1 changed file with 1 addition and 3 deletions.
diff --git a/faust/tables/base.py b/faust/tables/base.py
@@ -365,9 +365,7 @@ async def _del_old_keys(self) -> None:
         window = cast(WindowT, self.window)
         assert window
         for partition, timestamps in self._partition_timestamps.items():
-            while timestamps and window.stale(
-                timestamps[0], self._partition_latest_timestamp[partition]
-            ):
+            while timestamps and window.stale(timestamps[0], time.time()):
                 timestamp = heappop(timestamps)
                 keys_to_remove = self._partition_timestamp_keys.pop(
                     (partition, timestamp), None