fix(aiokafka): keep Faust assignor for changelog tables and update admin request compatibility

Ensure worker consumers use the Faust assignor whenever changelog tables are present,
even with zero standby replicas, so table recovery mapping remains correct. Also update
metadata/create-topics request construction for newer aiokafka APIs.

Author: Marco Moser

faust/__init__.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 """Python Stream processing."""
+
 # :copyright: (c) 2017-2020, Robinhood Markets, Inc.
 #             All rights reserved.
 # :license:   BSD (3 Clause), see LICENSE for more details.

faust/stores/aerospike.py

@@ -98,7 +98,7 @@ def _get(self, key: bytes) -> Optional[bytes]:
         key = (self.namespace, self.table_name, key)
         fun = self.client.get
         try:
-            (key, meta, bins) = self.aerospike_fun_call_with_retry(fun=fun, key=key)
+            key, meta, bins = self.aerospike_fun_call_with_retry(fun=fun, key=key)
             if bins:
                 return bins[self.BIN_KEY]
             return None
@@ -173,7 +173,7 @@ def _itervalues(self) -> Iterator[bytes]:
                 fun=fun, namespace=self.namespace, set=self.table_name
             )
             for result in scan.results():
-                (key, meta, bins) = result
+                key, meta, bins = result
                 if bins:
                     yield bins[self.BIN_KEY]
                 else:
@@ -193,8 +193,8 @@ def _iteritems(self) -> Iterator[Tuple[bytes, bytes]]:
                 fun=fun, namespace=self.namespace, set=self.table_name
             )
             for result in scan.results():
-                (key_data, meta, bins) = result
-                (ns, set, policy, key) = key_data
+                key_data, meta, bins = result
+                ns, set, policy, key = key_data
 
                 if bins:
                     bins = bins[self.BIN_KEY]
@@ -214,7 +214,7 @@ def _contains(self, key: bytes) -> bool:
         try:
             if self.app.conf.store_check_exists:
                 key = (self.namespace, self.table_name, key)
-                (key, meta) = self.aerospike_fun_call_with_retry(
+                key, meta = self.aerospike_fun_call_with_retry(
                     fun=self.client.exists, key=key
                 )
                 if meta:

faust/transport/drivers/aiokafka.py

@@ -27,9 +27,6 @@
 import aiokafka
 import aiokafka.abc
 import opentracing
-from packaging.version import Version
-
-_AIOKAFKA_HAS_API_VERSION = Version(aiokafka.__version__) < Version("0.13.0")
 from aiokafka import TopicPartition
 from aiokafka.consumer.group_coordinator import OffsetCommitRequest
 from aiokafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
@@ -45,7 +42,7 @@
 )
 from aiokafka.partitioner import DefaultPartitioner, murmur2
 from aiokafka.protocol.admin import CreateTopicsRequest
-from aiokafka.protocol.metadata import MetadataRequest_v1
+from aiokafka.protocol.metadata import MetadataRequest
 from aiokafka.structs import OffsetAndMetadata, TopicPartition as _TopicPartition
 from aiokafka.util import parse_kafka_version
 from mode import Service, get_logger
@@ -55,6 +52,7 @@
 from mode.utils.objects import cached_property
 from mode.utils.times import Seconds, humanize_seconds_ago, want_seconds
 from opentracing.ext import tags
+from packaging.version import Version
 from yarl import URL
 
 from faust.auth import (
@@ -96,6 +94,8 @@
 
 logger = get_logger(__name__)
 
+_AIOKAFKA_HAS_API_VERSION = Version(aiokafka.__version__) < Version("0.13.0")
+
 DEFAULT_GENERATION_ID = OffsetCommitRequest.DEFAULT_GENERATION_ID
 
 TOPIC_LENGTH_MAX = 249
@@ -511,9 +511,13 @@ def _create_worker_consumer(
         conf = self.app.conf
         if self.consumer.in_transaction:
             isolation_level = "read_committed"
+        # Table recovery depends on app.assignor state to map changelog
+        # active/standby partitions. Keep Faust assignor enabled whenever
+        # this app has changelog tables configured.
+        has_changelog_tables = bool(self.app.tables.changelog_topics)
         self._assignor = (
             self.app.assignor
-            if self.app.conf.table_standby_replicas > 0
+            if self.app.conf.table_standby_replicas > 0 or has_changelog_tables
             else RoundRobinPartitionAssignor
         )
         auth_settings = credentials_to_aiokafka_auth(
@@ -1513,7 +1517,7 @@ async def _get_controller_node(
         for node_id in nodes:
             if node_id is None:
                 raise NotReady("Not connected to Kafka Broker")
-            request = MetadataRequest_v1([])
+            request = MetadataRequest([])
             wait_result = await owner.wait(
                 client.send(node_id, request),
                 timeout=timeout,
@@ -1546,7 +1550,6 @@ async def _really_create_topic(
             owner.log.debug("Topic %r exists, skipping creation.", topic)
             return
 
-        protocol_version = 1
         extra_configs = config or {}
         config = self._topic_config(retention, compacting, deleting)
         config.update(extra_configs)
@@ -1563,7 +1566,7 @@ async def _really_create_topic(
             else:
                 raise Exception("Controller node is None")
 
-        request = CreateTopicsRequest[protocol_version](
+        request = CreateTopicsRequest(
             [(topic, partitions, replication, [], list(config.items()))],
             timeout,
             False,

tests/unit/transport/drivers/test_aiokafka.py

@@ -17,10 +17,10 @@
 import faust
 from faust import auth
 from faust.exceptions import ImproperlyConfigured, NotReady
-from faust.transport.drivers.aiokafka import _AIOKAFKA_HAS_API_VERSION
 from faust.sensors.monitor import Monitor
 from faust.transport.drivers import aiokafka as mod
 from faust.transport.drivers.aiokafka import (
+    _AIOKAFKA_HAS_API_VERSION,
     SLOW_PROCESSING_CAUSE_AGENT,
     SLOW_PROCESSING_CAUSE_STREAM,
     SLOW_PROCESSING_EXPLAINED,
@@ -796,6 +796,26 @@ def test__create_worker_consumer__transaction(self, *, cthread, app):
             isolation_level="read_committed",
         )
 
+    def test__create_worker_consumer__uses_roundrobin_without_tables(
+        self, *, cthread, app
+    ):
+        app.conf.table_standby_replicas = 0
+        app.tables._changelogs.clear()
+        transport = cthread.transport
+        with patch("aiokafka.AIOKafkaConsumer"):
+            cthread._create_worker_consumer(transport)
+        assert cthread._assignor is mod.RoundRobinPartitionAssignor
+
+    def test__create_worker_consumer__uses_faust_assignor_with_changelog_topics(
+        self, *, cthread, app
+    ):
+        app.conf.table_standby_replicas = 0
+        app.tables._changelogs["app-foo-changelog"] = Mock(name="table")
+        transport = cthread.transport
+        with patch("aiokafka.AIOKafkaConsumer"):
+            cthread._create_worker_consumer(transport)
+        assert cthread._assignor is app.assignor
+
     def assert_create_worker_consumer(
         self,
         cthread,