diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index bd3e85eca..7a3728f18 100644 --- a/.github/workflows/python-package.yml +++ b/.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: diff --git a/README.md b/README.md index c1391f2b6..e4212211a 100644 --- a/README.md +++ b/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 diff --git a/docs/includes/faq.txt b/docs/includes/faq.txt index 7641cb64a..1b6babd2b 100644 --- a/docs/includes/faq.txt +++ b/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). diff --git a/docs/introduction.rst b/docs/introduction.rst index 5932baa23..ad064a4f3 100644 --- a/docs/introduction.rst +++ b/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. diff --git a/docs/userguide/application.rst b/docs/userguide/application.rst index 832739c81..29c21b970 100644 --- a/docs/userguide/application.rst +++ b/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 diff --git a/examples/django/setup.py b/examples/django/setup.py index eda2bfd42..6cef929b8 100644 --- a/examples/django/setup.py +++ b/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'), diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py index 1a2a83454..3ebaf6b45 100644 --- a/faust/transport/drivers/aiokafka.py +++ b/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, diff --git a/pyproject.toml b/pyproject.toml index 6ddfebde8..b4b7f923c 100644 --- a/pyproject.toml +++ b/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", diff --git a/setup.py b/setup.py index 6608fba62..390cad0a9 100644 --- a/setup.py +++ b/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"), diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py index fc54fbcfa..d36665704 100644 --- a/tests/unit/transport/drivers/test_aiokafka.py +++ b/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, diff --git a/tox.ini b/tox.ini index 59aff06e1..e1673e689 100644 --- a/tox.ini +++ b/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 =