Cancel getmany if generation_id changes (faust-streaming#234)

* Cancel getmany if generation_id changes

* black

Co-authored-by: Eric Kerstens <ekerstens@expediagroup.com>

faust/transport/consumer.py

@@ -701,6 +701,7 @@ async def getmany(self, timeout: float) -> AsyncIterator[Tuple[TP, Message]]:
         # has 1 partition, then t2 will end up being starved most of the time.
         #
         # We solve this by going round-robin through each topic.
+        generation_id = self.app.consumer_generation_id
         records, active_partitions = await self._wait_next_records(timeout)
         if records is None or self.should_stop:
             return
@@ -711,6 +712,14 @@ async def getmany(self, timeout: float) -> AsyncIterator[Tuple[TP, Message]]:
             for tp, record in records_it:
                 if not self.flow_active:
                     break
+                new_generation_id = self.app.consumer_generation_id
+                if new_generation_id != generation_id:
+                    self.log.dev(
+                        "Generation id changed from %r to %r. Cancelling getmany.",
+                        generation_id,
+                        new_generation_id,
+                    )
+                    break
                 if (
                     active_partitions is None
                     or tp in active_partitions

faust/types/models.py

@@ -57,7 +57,6 @@ def __init_subclass__(
     class _UsingKwargsInNew(_InitSubclassCheck, ident=909):
         ...
 
-
 except TypeError:
     abc_compatible_with_init_subclass = False
 else:

faust/utils/json.py

@@ -179,7 +179,6 @@ def loads(s: str, json_loads: Callable = orjson.loads, **kwargs: Any) -> Any:
         """Deserialize json string."""
         return json_loads(s)
 
-
 else:
 
     def dumps(