Fix for rebalance Exception causing recovery to crash (#57)

* Fix #51

* fix for crash on _commit due to rebalance

faust/tables/recovery.py

@@ -18,6 +18,7 @@
 )
 
 import opentracing
+from kafka.errors import IllegalStateError
 from mode import Service, get_logger
 from mode.services import WaitArgT
 from mode.utils.times import humanize_seconds, humanize_seconds_ago
@@ -508,6 +509,11 @@ async def _restart_recovery(self) -> None:
                 for _span in spans:
                     finish_span(_span, error=exc)
                 continue  # another rebalance started
+            except IllegalStateError as exc:
+                self.log.dev("RAISED REBALANCE AGAIN")
+                for _span in spans:
+                    finish_span(_span, error=exc)
+                continue  # another rebalance started
             except ServiceStopped as exc:
                 self.log.dev("RAISED SERVICE STOPPED")
                 for _span in spans:

faust/transport/drivers/aiokafka.py

@@ -526,21 +526,18 @@ async def _commit(self, offsets: Mapping[TP, int]) -> bool:
                 return False
             self.log.exception("Committing raised exception: %r", exc)
             await self.crash(exc)
-            self.supervisor.wakeup()
             return False
         except IllegalStateError as exc:
             self.log.exception(
                 "Got exception: %r\nCurrent assignment: %r", exc, self.assignment()
             )
             await self.crash(exc)
-            self.supervisor.wakeup()
             return False
         except Exception as exc:
             self.log.exception(
                 "Got exception: %r\nCurrent assignment: %r", exc, self.assignment()
             )
             await self.crash(exc)
-            self.supervisor.wakeup()
             return False
         return True