fix: support aiokafka 0.13.0 which removed api_version parameter (#674)

aiokafka 0.13.0 removed the `api_version` parameter from both
`AIOKafkaConsumer` and `AIOKafkaProducer`. Since faust requires
`aiokafka>=0.10.0`, it needs to handle both old and new versions.

This adds a version check and conditionally includes `api_version`
only when running with aiokafka < 0.13.0.

Fixes #672

Co-authored-by: nightcityblade <[email protected]>

Author: nightcityblade

faust/transport/drivers/aiokafka.py

@@ -27,6 +27,9 @@
 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
@@ -529,8 +532,11 @@ def _create_worker_consumer(
                 f"broker_request_timeout={request_timeout}"
             )
 
+        consumer_kwargs: dict[str, Any] = {}
+        if _AIOKAFKA_HAS_API_VERSION:
+            consumer_kwargs["api_version"] = conf.consumer_api_version
         return aiokafka.AIOKafkaConsumer(
-            api_version=conf.consumer_api_version,
+            **consumer_kwargs,
             client_id=conf.broker_client_id,
             group_id=conf.id,
             group_instance_id=conf.consumer_group_instance_id,
@@ -1111,7 +1117,7 @@ def __post_init__(self) -> None:
 
     def _settings_default(self) -> Mapping[str, Any]:
         transport = cast(Transport, self.transport)
-        return {
+        settings: dict[str, Any] = {
             "bootstrap_servers": server_list(transport.url, transport.default_port),
             "client_id": self.client_id,
             "acks": self.acks,
@@ -1122,10 +1128,12 @@ def _settings_default(self) -> Mapping[str, Any]:
             "security_protocol": "SSL" if self.ssl_context else "PLAINTEXT",
             "partitioner": self.partitioner,
             "request_timeout_ms": int(self.request_timeout * 1000),
-            "api_version": self._api_version,
             "metadata_max_age_ms": self.app.conf.producer_metadata_max_age_ms,
             "connections_max_idle_ms": self.app.conf.producer_connections_max_idle_ms,
         }
+        if _AIOKAFKA_HAS_API_VERSION:
+            settings["api_version"] = self._api_version
+        return settings
 
     def _settings_auth(self) -> Mapping[str, Any]:
         return credentials_to_aiokafka_auth(self.credentials, self.ssl_context)

tests/unit/transport/drivers/test_aiokafka.py

@@ -17,6 +17,7 @@
 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 (
@@ -813,8 +814,7 @@ def assert_create_worker_consumer(
             c = cthread._create_worker_consumer(transport)
             assert c is AIOKafkaConsumer.return_value
             max_poll_interval = conf.broker_max_poll_interval
-            AIOKafkaConsumer.assert_called_once_with(
-                api_version=app.conf.consumer_api_version,
+            expected_kwargs = dict(
                 client_id=conf.broker_client_id,
                 group_id=conf.id,
                 group_instance_id=conf.consumer_group_instance_id,
@@ -841,6 +841,9 @@ def assert_create_worker_consumer(
                 # flush_spans=cthread.flush_spans,
                 **auth_settings,
             )
+            if _AIOKAFKA_HAS_API_VERSION:
+                expected_kwargs["api_version"] = app.conf.consumer_api_version
+            AIOKafkaConsumer.assert_called_once_with(**expected_kwargs)
 
     def test__create_client_consumer(self, *, cthread, app):
         transport = cthread.transport
@@ -1382,7 +1385,7 @@ def assert_new_producer(
         with patch("aiokafka.AIOKafkaProducer") as AIOKafkaProducer:
             p = producer._new_producer()
             assert p is AIOKafkaProducer.return_value
-            AIOKafkaProducer.assert_called_once_with(
+            expected_kwargs = dict(
                 bootstrap_servers=bootstrap_servers,
                 client_id=client_id,
                 acks=acks,
@@ -1393,12 +1396,14 @@ def assert_new_producer(
                 security_protocol=security_protocol,
                 partitioner=producer.partitioner,
                 transactional_id=None,
-                api_version=api_version,
                 metadata_max_age_ms=metadata_max_age_ms,
                 connections_max_idle_ms=connections_max_idle_ms,
                 request_timeout_ms=request_timeout_ms,
                 **kwargs,
             )
+            if _AIOKAFKA_HAS_API_VERSION:
+                expected_kwargs["api_version"] = api_version
+            AIOKafkaProducer.assert_called_once_with(**expected_kwargs)
 
 
 class TestProducer(ProducerBaseTest):
@@ -1475,7 +1480,13 @@ def test__new_producer(self, *, app):
         [
             pytest.param(
                 {"api_version": "auto"},
-                marks=pytest.mark.conf(producer_api_version="auto"),
+                marks=[
+                    pytest.mark.conf(producer_api_version="auto"),
+                    pytest.mark.skipif(
+                        not _AIOKAFKA_HAS_API_VERSION,
+                        reason="api_version removed in aiokafka>=0.13.0",
+                    ),
+                ],
             ),
             pytest.param({"acks": -1}, marks=pytest.mark.conf(producer_acks="all")),
             pytest.param(