diff --git a/CHANGELOG.md b/CHANGELOG.md index ebeab99ee..409b8843c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,9 @@ resumes the Keep a Changelog format. - New userguide page: *FastAPI and other ASGI applications*. ### Fixed +- Support aiokafka 0.15.0, which removed the internal + `aiokafka.util.parse_kafka_version` helper used to determine Kafka header + support. - Faust apps no longer resolve an event loop when agents, tables or the transport are declared at import time. Previously that pinned the app to a loop that was never run, so starting it from `asyncio.run()` — as uvicorn diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py index 544dbd483..40daa1508 100644 --- a/faust/transport/drivers/aiokafka.py +++ b/faust/transport/drivers/aiokafka.py @@ -43,7 +43,6 @@ from aiokafka.protocol.admin import CreateTopicsRequest 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 from mode.threads import ServiceThread, WorkerThread from mode.utils import text @@ -1183,8 +1182,7 @@ def __post_init__(self) -> None: if self.partitioner is None: self.partitioner = DefaultPartitioner() if self._api_version != "auto": - wanted_api_version = parse_kafka_version(self._api_version) - if wanted_api_version < (0, 11): + if Version(self._api_version) < Version("0.11"): self.allow_headers = False def _settings_default(self) -> Mapping[str, Any]: