allow only positive offsets to dictate recovery (#100)

faust/tables/recovery.py

@@ -751,7 +751,7 @@ async def _maybe_signal_recovery_end(timeout=False, timeout_count=0) -> None:
             # the aiokafka consumer position and draining of the queue
             if timeout and self.app.in_transaction and timeout_count > 1:
                 await detect_aborted_tx()
-            if not self.active_remaining_total() and self.in_recovery:
+            if not self.need_recovery() and self.in_recovery:
                 # apply anything stuck in the buffers
                 self.flush_buffers()
                 self._set_recovery_ended()