fix(aiokafka): keep Faust assignor for changelog tables; drop Python 3.9

.github/workflows/python-package.yml

@@ -42,7 +42,7 @@ jobs:
       # for example if a test fails only when Cython is enabled
       fail-fast: false
       matrix:
-        python-version: ['3.9', '3.10', '3.11', '3.12', '3.13']
+        python-version: ['3.10', '3.11', '3.12', '3.13']
         use-cython: ['true', 'false']
         experimental: [false]
     env:

README.md

@@ -416,7 +416,7 @@ Yes! Use the `asyncio` reactor implementation: https://twistedmatrix.com/documen
 
 ### Will you support Python 2.7 or Python 3.5
 
-No. Faust requires Python 3.8 or later, since it heavily uses features that were
+No. Faust requires Python 3.10 or later, since it heavily uses features that were
 introduced in Python 3.6 (`async`, `await`, variable type annotations).
 
 ### I get a maximum number of open files exceeded error by RocksDB when running a Faust app locally. How can I fix this

docs/includes/faq.txt

@@ -55,7 +55,7 @@ https://twistedmatrix.com/documents/17.1.0/api/twisted.internet.asyncioreactor.h
 Will you support Python 2.7 or Python 3.5?
 ------------------------------------------
 
-No. Faust requires Python 3.8 or later, since it heavily uses features that were
+No. Faust requires Python 3.10 or later, since it heavily uses features that were
 introduced in Python 3.6 (`async`, `await`, variable type annotations).
 
 

docs/introduction.rst

@@ -162,7 +162,7 @@ What do I need?
 
     - RocksDB 5.0 or later, :pypi:`python-rocksdb`
 
-Faust requires Python 3.8 or later, and a running Kafka broker.
+Faust requires Python 3.10 or later, and a running Kafka broker.
 
 There's no plan to support earlier Python versions.
 Please get in touch if this is something you want to work on.

docs/userguide/application.rst

@@ -1374,7 +1374,7 @@ setuptools to install a command-line program for your project.
             include_package_data=True,
             zip_safe=False,
             install_requires=['faust'],
-            python_requires='~=3.8',
+            python_requires='~=3.10',
         )
 
     For inspiration you can also look to the `setup.py` files in the

examples/django/setup.py

@@ -95,7 +95,7 @@ def reqs(*f):
     license='BSD',
     packages=find_packages(exclude=['ez_setup', 'tests', 'tests.*']),
     include_package_data=True,
-    python_requires='>=3.8.0',
+    python_requires='>=3.10.0',
     keywords=[],
     zip_safe=False,
     install_requires=reqs('default.txt'),

faust/transport/drivers/aiokafka.py

@@ -42,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, MetadataRequest_v1
 from aiokafka.structs import OffsetAndMetadata, TopicPartition as _TopicPartition
 from aiokafka.util import parse_kafka_version
 from mode import Service, get_logger
@@ -86,8 +86,6 @@
 from faust.types.transports import ConsumerT, PartitionerT, ProducerT
 from faust.utils.tracing import noop_span, set_current_span, traced_from_parent_span
 
-_AIOKAFKA_HAS_API_VERSION = Version(aiokafka.__version__) < Version("0.13.0")
-
 __all__ = ["Consumer", "Producer", "Transport"]
 
 # if not hasattr(aiokafka, '__robinhood__'):  # pragma: no cover
@@ -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,11 @@ 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([])
+            if _AIOKAFKA_HAS_API_VERSION:
+                # aiokafka < 0.13: MetadataRequest is a versioned list.
+                request = MetadataRequest_v1([])
+            else:
+                request = MetadataRequest([])
             wait_result = await owner.wait(
                 client.send(node_id, request),
                 timeout=timeout,
@@ -1546,7 +1554,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,11 +1570,16 @@ async def _really_create_topic(
             else:
                 raise Exception("Controller node is None")
 
-        request = CreateTopicsRequest[protocol_version](
+        create_topics_args = (
             [(topic, partitions, replication, [], list(config.items()))],
             timeout,
             False,
         )
+        if _AIOKAFKA_HAS_API_VERSION:
+            # aiokafka < 0.13: CreateTopicsRequest is indexed by protocol version.
+            request = CreateTopicsRequest[1](*create_topics_args)
+        else:
+            request = CreateTopicsRequest(*create_topics_args)
         wait_result = await owner.wait(
             client.send(controller_node, request),
             timeout=timeout,

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "faust-streaming"
 description = "Python Stream Processing. A Faust fork"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 dynamic = [
     "version",
     "optional-dependencies",
@@ -26,8 +26,6 @@ classifiers = [
     "License :: OSI Approved :: BSD License",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",

setup.py

@@ -47,8 +47,8 @@
 LIBRARIES = []
 E_UNSUPPORTED_PYTHON = NAME + " 1.0 requires Python %%s or later!"
 
-if sys.version_info < (3, 8):
-    raise Exception(E_UNSUPPORTED_PYTHON % ("3.8",))  # NOQA
+if sys.version_info < (3, 10):
+    raise Exception(E_UNSUPPORTED_PYTHON % ("3.10",))  # NOQA
 
 from pathlib import Path  # noqa
 
@@ -198,7 +198,7 @@ def do_setup(**kwargs):
         # PEP-561: https://www.python.org/dev/peps/pep-0561/
         package_data={"faust": ["py.typed"]},
         include_package_data=True,
-        python_requires=">=3.8.0",
+        python_requires=">=3.10.0",
         zip_safe=False,
         install_requires=reqs("requirements.txt"),
         tests_require=reqs("test.txt"),

tests/unit/transport/drivers/test_aiokafka.py

@@ -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,

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = 3.12,3.11,3.10,3.9,3.8,flake8,apicheck,configcheck,typecheck,docstyle,bandit,spell
+envlist = 3.12,3.11,3.10,flake8,apicheck,configcheck,typecheck,docstyle,bandit,spell
 
 [testenv]
 deps=
@@ -20,8 +20,6 @@ basepython =
     3.12,flake8,apicheck,linkcheck,configcheck,typecheck,docstyle,bandit,spell: python3.12
     3.11,flake8,apicheck,linkcheck,configcheck,typecheck,docstyle,bandit,spell: python3.11
     3.10,flake8,apicheck,linkcheck,configcheck,typecheck,docstyle,bandit,spell: python3.10
-    3.9,flake8,apicheck,linkcheck,configcheck,typecheck,docstyle,bandit,spell: python3.9
-    3.8,flake8,apicheck,linkcheck,configcheck,typecheck,docstyle,bandit,spell: python3.8
 
 [testenv:apicheck]
 setenv =