From 2c7cc1e00db4a9b29fe47f8343b091105423a9d4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 15:14:53 +0000 Subject: [PATCH 1/2] Add ClientConfig, converting Kafka client configs via kafkaesq Faust deals with three spellings of the same Kafka client configuration: aiokafka's constructor kwargs (session_timeout_ms), librdkafka's dotted keys (session.timeout.ms), and app settings (broker_session_timeout, in seconds). Moving a config between them is manual work today. Add kafkaesq as an optional dependency -- pip install "faust-streaming[kafkaesq]" -- and a ClientConfig class that holds one configuration and hands it out in whichever spelling is wanted: config = ClientConfig.from_confluent(client_properties) consumer = AIOKafkaConsumer('topic', **config.as_aiokafka()) settings = config.as_app_settings() app = faust.App(settings.pop('id'), **settings) and the other way round, from_app/from_app_settings turn an app's settings into config for a plain client -- from_app taking the brokers from the app's transport, so the client connects where the app does. Values are clamped to the ranges librdkafka enforces on the way out, since Faust's settings are not bounded by them: producer_request_timeout defaults to 20 minutes, above librdkafka's 15-minute ceiling for request.timeout.ms, which would fail client construction outright. Nothing in Faust depends on this: the transports configure themselves from app settings exactly as before, and neither needs kafkaesq installed. Without it the class raises ImproperlyConfigured with the install hint. Authentication is out of scope -- Faust configures it with a broker_credentials object built at runtime, which no config file can describe, so security keys are reported as unmapped when converting to app settings. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BYhhm6H1FQskUmDqjrft7G --- docs/includes/installation.txt | 6 + docs/includes/settingref.txt | 55 +++ docs/reference/faust.transport.kafkaesq.rst | 11 + docs/reference/index.rst | 1 + faust/transport/kafkaesq.py | 349 ++++++++++++++++++++ requirements/extras/kafkaesq.txt | 1 + requirements/test.txt | 1 + setup.py | 1 + tests/unit/transport/test_kafkaesq.py | 267 +++++++++++++++ 9 files changed, 692 insertions(+) create mode 100644 docs/reference/faust.transport.kafkaesq.rst create mode 100644 faust/transport/kafkaesq.py create mode 100644 requirements/extras/kafkaesq.txt create mode 100644 tests/unit/transport/test_kafkaesq.py diff --git a/docs/includes/installation.txt b/docs/includes/installation.txt index a44f485d6..c0b4034e9 100644 --- a/docs/includes/installation.txt +++ b/docs/includes/installation.txt @@ -57,6 +57,12 @@ Transports for using the :pypi:`confluent-kafka` client through the ``confluent://`` transport. +:``faust[kafkaesq]``: + for converting Kafka client configuration between the spellings + :pypi:`aiokafka`, :pypi:`confluent-kafka` and Faust app settings use -- + turning an existing librdkafka config into app settings, or app settings + into arguments for a plain client. See :ref:`client-config-conversion`. + Codecs ~~~~~~ diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 669894cff..65ff3fe82 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -399,6 +399,61 @@ You can also pass a list of URLs: suitable for tables), and do not create any necessary internal topics (you have to create them manually). +.. _client-config-conversion: + +**Converting client configuration** + +The clients behind these transports spell their configuration differently: +:pypi:`aiokafka` takes ``session_timeout_ms``, librdkafka takes +``session.timeout.ms``, and app settings are a third spelling again +(:setting:`broker_session_timeout`, in seconds). Install the ``kafkaesq`` +bundle: + +.. sourcecode:: console + + $ pip install "faust-streaming[kafkaesq]" + +and :class:`faust.transport.kafkaesq.ClientConfig` converts between them. A +librdkafka config -- a Confluent Cloud ``client.properties``, say -- becomes +arguments for :class:`faust.App`: + +.. sourcecode:: python + + from faust.transport.kafkaesq import ClientConfig + + config = ClientConfig.from_confluent({ + 'bootstrap.servers': 'kafka.example.com:9092', + 'group.id': 'billing', + 'session.timeout.ms': 45000, + }) + + settings = config.as_app_settings() + app = faust.App(settings.pop('id'), **settings) + +and, the other way round, an app's settings configure a plain client that +talks to the same brokers: + +.. sourcecode:: python + + from faust.transport.kafkaesq import PRODUCER_SETTINGS, ClientConfig + + config = ClientConfig.from_app(app, PRODUCER_SETTINGS) + producer = confluent_kafka.Producer(config.as_confluent()) + # ... or the same settings as aiokafka kwargs + producer = AIOKafkaProducer(**config.as_aiokafka()) + +This is a conversion helper, not something the transports use: both configure +themselves from your app settings as they always have, and neither needs +:pypi:`kafkaesq` installed. + +.. note:: + + Authentication is not converted. Faust configures it with a + :setting:`broker_credentials` object built at runtime, which no config + file can describe, so ``security.protocol``, ``sasl.*`` and ``ssl.*`` + keys are reported as unmapped when converting to app settings -- set + :setting:`broker_credentials` yourself. + .. setting:: broker_credentials diff --git a/docs/reference/faust.transport.kafkaesq.rst b/docs/reference/faust.transport.kafkaesq.rst new file mode 100644 index 000000000..a2d7605c2 --- /dev/null +++ b/docs/reference/faust.transport.kafkaesq.rst @@ -0,0 +1,11 @@ +===================================================== + ``faust.transport.kafkaesq`` +===================================================== + +.. contents:: + :local: +.. currentmodule:: faust.transport.kafkaesq + +.. automodule:: faust.transport.kafkaesq + :members: + :undoc-members: diff --git a/docs/reference/index.rst b/docs/reference/index.rst index 979616bda..9bdb3a2e6 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -159,6 +159,7 @@ Transports faust.transport.producer faust.transport.drivers faust.transport.drivers.aiokafka + faust.transport.kafkaesq faust.transport.utils Assignor diff --git a/faust/transport/kafkaesq.py b/faust/transport/kafkaesq.py new file mode 100644 index 000000000..a03df33bc --- /dev/null +++ b/faust/transport/kafkaesq.py @@ -0,0 +1,349 @@ +"""Kafka client configuration, convertible between client spellings. + +Faust's two transports talk to Kafka through clients that spell their +configuration differently: the ``kafka://``/``aiokafka://`` transport uses +:pypi:`aiokafka` constructor kwargs (``session_timeout_ms``), the +``confluent://`` transport uses librdkafka's dotted keys +(``session.timeout.ms``), and app settings are a third spelling again +(:setting:`broker_session_timeout`, in seconds). + +:class:`ClientConfig` holds one configuration and hands it out in whichever +of those spellings is wanted, so a config written for one client can drive +the other -- or a Faust app:: + + from faust.transport.kafkaesq import ClientConfig + + config = ClientConfig.from_confluent({ + 'bootstrap.servers': 'localhost:9092', + 'group.id': 'billing', + 'session.timeout.ms': 45000, + }) + + consumer = AIOKafkaConsumer('topic', **config.as_aiokafka()) + + settings = config.as_app_settings() + app = faust.App(settings.pop('id'), **settings) + +The conversion tables are :pypi:`kafkaesq`'s, installed with the ``kafkaesq`` +bundle:: + + $ pip install "faust-streaming[kafkaesq]" + +It is an optional dependency and nothing in Faust needs it: the transports +configure themselves from app settings as they always have. Without it +:data:`HAS_KAFKAESQ` is :const:`False` and :class:`ClientConfig` raises +:exc:`~faust.exceptions.ImproperlyConfigured`. + +Note: + Authentication is out of scope here. Faust configures it with a + :setting:`broker_credentials` object built at runtime, which no config + file can describe, so ``security.protocol``, ``sasl.*`` and ``ssl.*`` + keys have no app setting and are reported as unmapped by + :meth:`~ClientConfig.as_app_settings`. +""" + +import typing +from collections.abc import Mapping as _AbcMapping +from typing import Any, Dict, Iterable, Mapping, Optional, Tuple, Union + +from mode import get_logger + +from faust.exceptions import ImproperlyConfigured + +try: + import kafkaesq +except ImportError: # pragma: no cover + kafkaesq = None # type: ignore[assignment] + +if typing.TYPE_CHECKING: + from faust.types.settings import Settings as _Settings +else: + + class _Settings: ... # noqa + + +__all__ = [ + "HAS_KAFKAESQ", + "CONFLUENT_LIMITS", + "CONSUMER_SETTINGS", + "PRODUCER_SETTINGS", + "ClientConfig", + "require_kafkaesq", +] + +logger = get_logger(__name__) + +#: Set when :pypi:`kafkaesq` is installed (the ``faust[kafkaesq]`` bundle). +HAS_KAFKAESQ: bool = kafkaesq is not None + +#: App settings that describe a consumer, for :meth:`ClientConfig.from_app`. +#: +#: :setting:`broker` is absent: which brokers to connect to is the transport +#: URL's job, and :meth:`~ClientConfig.from_app` takes it from there. +#: :setting:`broker_request_timeout` is absent too, because librdkafka's +#: ``request.timeout.ms`` is a producer property that a consumer instance +#: warns about and ignores. +CONSUMER_SETTINGS: Tuple[str, ...] = ( + "id", + "broker_client_id", + "broker_commit_interval", + "broker_session_timeout", + "broker_heartbeat_interval", + "broker_max_poll_interval", + "broker_check_crcs", + "consumer_auto_offset_reset", + "consumer_group_instance_id", + "consumer_max_fetch_size", + "consumer_metadata_max_age_ms", + "consumer_connections_max_idle_ms", +) + +#: App settings that describe a producer, for :meth:`ClientConfig.from_app`. +PRODUCER_SETTINGS: Tuple[str, ...] = ( + "broker_client_id", + "producer_request_timeout", + "producer_acks", + "producer_compression_type", + "producer_linger", + "producer_max_request_size", + "producer_metadata_max_age_ms", + "producer_connections_max_idle_ms", +) + +#: Ranges librdkafka enforces, for :meth:`ClientConfig.as_confluent`. A value +#: outside them fails client construction outright, and Faust's settings are +#: not bounded by them: :setting:`producer_request_timeout` defaults to 20 +#: minutes, where ``request.timeout.ms`` stops at 15. +CONFLUENT_LIMITS: Mapping[str, Tuple[int, int]] = { + "request.timeout.ms": (1, 900000), + "session.timeout.ms": (1, 3600000), + "heartbeat.interval.ms": (1, 3600000), + "max.poll.interval.ms": (1, 86400000), + "metadata.max.age.ms": (1, 86400000), + "auto.commit.interval.ms": (0, 86400000), + "connections.max.idle.ms": (0, 2147483647), + "max.partition.fetch.bytes": (1, 1000000000), + "message.max.bytes": (1000, 1000000000), + "linger.ms": (0, 900000), +} + + +def require_kafkaesq() -> Any: + """Return the :pypi:`kafkaesq` module, raising if it is not installed. + + Raises: + ~faust.exceptions.ImproperlyConfigured: when the library is missing. + """ + if kafkaesq is None: + raise ImproperlyConfigured( + "Converting Kafka client configs requires the kafkaesq library: " + 'pip install "faust-streaming[kafkaesq]"' + ) + return kafkaesq + + +class ClientConfig: + """One Kafka client configuration, in every spelling Faust deals with. + + Held internally as a librdkafka config -- the spelling :pypi:`kafkaesq` + converts through -- and read back with :meth:`as_confluent`, + :meth:`as_aiokafka` or :meth:`as_app_settings`. + + Build one with :meth:`from_confluent`, :meth:`from_aiokafka` or + :meth:`from_app` rather than calling the constructor, unless you already + have a librdkafka config in hand. + + Raises: + ~faust.exceptions.ImproperlyConfigured: when :pypi:`kafkaesq` is not + installed. + """ + + #: The configuration, in librdkafka's dotted-key spelling. + config: Dict[str, Any] + + def __init__(self, config: Optional[Mapping[str, Any]] = None) -> None: + require_kafkaesq() + self.config = dict(config or {}) + + def __repr__(self) -> str: + return f"<{type(self).__name__}: {sorted(self.config)}>" + + def __eq__(self, other: Any) -> bool: + if isinstance(other, ClientConfig): + return self.config == other.config + return NotImplemented + + @classmethod + def from_confluent(cls, config: Mapping[str, Any]) -> "ClientConfig": + """Build from a :pypi:`confluent_kafka` (librdkafka) config.""" + return cls(config) + + @classmethod + def from_aiokafka( + cls, + config: Optional[Mapping[str, Any]] = None, + *, + on_unmapped: str = "warn", + **kwargs: Any, + ) -> "ClientConfig": + """Build from :pypi:`aiokafka` constructor kwargs. + + Takes a mapping of kwargs, keyword arguments, or both. + + Arguments: + on_unmapped: What to do with kwargs that have no librdkafka + equivalent -- ``"warn"`` (the default; the kwarg is dropped), + ``"raise"`` or ``"ignore"``. An ``ssl_context`` is one of + them: it is a Python object that cannot be turned back into + the certificate paths librdkafka wants. + """ + lib = require_kafkaesq() + return cls(lib.aiokafka_to_confluent(config, on_unmapped=on_unmapped, **kwargs)) + + @classmethod + def from_app_settings( + cls, + conf: Union[_Settings, Mapping[str, Any]], + keys: Iterable[str] = CONSUMER_SETTINGS, + *, + on_unmapped: str = "ignore", + ) -> "ClientConfig": + """Build from app settings. + + Arguments: + conf: App settings -- ``app.conf``, or a plain mapping of setting + names to values. + keys: Which settings to take, :data:`CONSUMER_SETTINGS` by + default; :data:`PRODUCER_SETTINGS` describes a producer. + Settings that are unset are skipped, leaving the client its + own default for them. + on_unmapped: What to do with settings that have no librdkafka + equivalent, ``"ignore"`` by default. Most app settings + describe Faust rather than its client, so ``"warn"`` is + noisy unless `keys` is a list you chose yourself. + + Note: + :setting:`broker` is not among :data:`CONSUMER_SETTINGS`: pass + ``bootstrap.servers`` yourself, or use :meth:`from_app`, which + takes it from the app's transport. + """ + lib = require_kafkaesq() + return cls( + lib.faust_to_confluent(_extract(conf, tuple(keys)), on_unmapped=on_unmapped) + ) + + @classmethod + def from_app( + cls, + app: Any, + keys: Iterable[str] = CONSUMER_SETTINGS, + *, + on_unmapped: str = "ignore", + ) -> "ClientConfig": + """Build from a running app's settings, brokers included. + + As :meth:`from_app_settings`, but ``bootstrap.servers`` is taken from + the app's transport, so the client this configures connects where the + app does:: + + config = ClientConfig.from_app(app, PRODUCER_SETTINGS) + producer = confluent_kafka.Producer(config.as_confluent()) + """ + config = cls.from_app_settings(app.conf, keys, on_unmapped=on_unmapped) + config.config["bootstrap.servers"] = _server_list(app.transport) + return config + + def as_confluent(self, *, clamp: bool = True) -> Dict[str, Any]: + """Return the config for a :pypi:`confluent_kafka` client. + + Arguments: + clamp: Bring values into the range librdkafka accepts for their + key (:data:`CONFLUENT_LIMITS`), logging each one changed. + Pass :const:`False` to get the values unaltered and have the + client reject them itself. + """ + config = dict(self.config) + return self._clamp(config) if clamp else config + + def as_aiokafka( + self, *, on_unmapped: str = "warn", **kwargs: Any + ) -> Dict[str, Any]: + """Return the config as :pypi:`aiokafka` constructor kwargs. + + librdkafka ``ssl.*`` file options are folded into a single + ``ssl_context``, which is the form aiokafka takes. + + Arguments: + on_unmapped: What to do with keys aiokafka has no kwarg for + (librdkafka internals, callbacks, ...): ``"warn"`` (the + default), ``"raise"`` or ``"ignore"``. + kwargs: Passed through to :pypi:`kafkaesq`, which understands + ``build_ssl_context=False`` for configs whose certificate + files are not present on this machine. + """ + lib = require_kafkaesq() + return lib.confluent_to_aiokafka(self.config, on_unmapped=on_unmapped, **kwargs) + + def as_app_settings(self, *, on_unmapped: str = "warn") -> Dict[str, Any]: + """Return the config as :class:`faust.App` keyword arguments. + + ``group.id`` becomes the app ``id`` and ``bootstrap.servers`` becomes + a :setting:`broker` URL list, so the result goes straight into an + app:: + + settings = config.as_app_settings() + app = faust.App(settings.pop('id'), **settings) + + Arguments: + on_unmapped: What to do with keys that have no app setting -- + authentication among them, see the module note: ``"warn"`` + (the default), ``"raise"`` or ``"ignore"``. + """ + lib = require_kafkaesq() + return lib.confluent_to_faust(self.config, on_unmapped=on_unmapped) + + @staticmethod + def _clamp(config: Dict[str, Any]) -> Dict[str, Any]: + for key, (low, high) in CONFLUENT_LIMITS.items(): + value = config.get(key) + if not isinstance(value, int) or isinstance(value, bool): + continue + if not low <= value <= high: + limit = low if value < low else high + logger.warning( + "confluent-kafka only accepts %s between %s and %s: " + "using %s instead of %s", + key, + low, + high, + limit, + value, + ) + config[key] = limit + return config + + +def _server_list(transport: Any) -> str: + """Spell a transport's broker URLs as ``bootstrap.servers``. + + The same host list the drivers build for their own clients. + """ + return ",".join( + f"{url.host or '127.0.0.1'}:{url.port or transport.default_port}" + for url in transport.url + ) + + +def _extract( + conf: Union[_Settings, Mapping[str, Any]], keys: Tuple[str, ...] +) -> Dict[str, Any]: + """Read `keys` out of an app settings object (or a plain mapping). + + Settings left unset are skipped: an absent key means "whatever the client + defaults to", which is what Faust's own transports do with them too. + """ + if isinstance(conf, _AbcMapping): + values = {key: conf.get(key) for key in keys if key in conf} + else: + values = {key: getattr(conf, key, None) for key in keys} + return {key: value for key, value in values.items() if value is not None} diff --git a/requirements/extras/kafkaesq.txt b/requirements/extras/kafkaesq.txt new file mode 100644 index 000000000..137dae43b --- /dev/null +++ b/requirements/extras/kafkaesq.txt @@ -0,0 +1 @@ +kafkaesq>=0.2.1 diff --git a/requirements/test.txt b/requirements/test.txt index 829398938..9413d9d0a 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -31,6 +31,7 @@ intervaltree # type-checks the faust package, so the lint job needs it installed. -r typecheck.txt -r extras/datadog.txt +-r extras/kafkaesq.txt -r extras/opentracing.txt -r extras/redis.txt -r extras/statsd.txt diff --git a/setup.py b/setup.py index 7a3190b70..b59c41bad 100644 --- a/setup.py +++ b/setup.py @@ -30,6 +30,7 @@ "datadog", "debug", "fast", + "kafkaesq", "opentracing", "orjson", "prometheus", diff --git a/tests/unit/transport/test_kafkaesq.py b/tests/unit/transport/test_kafkaesq.py new file mode 100644 index 000000000..7da11a737 --- /dev/null +++ b/tests/unit/transport/test_kafkaesq.py @@ -0,0 +1,267 @@ +"""Tests for :class:`faust.transport.kafkaesq.ClientConfig`. + +The class needs :pypi:`kafkaesq` (the ``faust[kafkaesq]`` bundle) and says so +when it is missing; that path is tested by faking the library away rather +than by skipping, so it is covered wherever the suite runs. +""" + +import ssl +from unittest.mock import patch + +import pytest + +import faust +from faust.exceptions import ImproperlyConfigured +from faust.transport import kafkaesq as mod +from faust.transport.kafkaesq import ( + CONSUMER_SETTINGS, + PRODUCER_SETTINGS, + ClientConfig, + require_kafkaesq, +) + +#: Skips the tests that need the real library. +needs_kafkaesq = pytest.mark.skipif( + not mod.HAS_KAFKAESQ, reason="kafkaesq is not installed" +) + +CONFLUENT_CONFIG = { + "bootstrap.servers": "h1:9092,h2:9092", + "group.id": "billing", + "client.id": "worker-1", + "session.timeout.ms": 45000, + "auto.offset.reset": "latest", +} + + +@pytest.fixture() +def no_kafkaesq(): + """Pretend the optional dependency is not installed.""" + with patch.object(mod, "kafkaesq", None): + yield + + +class Test_require_kafkaesq: + @needs_kafkaesq + def test_returns_the_module(self): + assert require_kafkaesq() is mod.kafkaesq + + def test_raises_when_missing(self, *, no_kafkaesq): + with pytest.raises(ImproperlyConfigured, match="faust-streaming\\[kafkaesq\\]"): + require_kafkaesq() + + def test_client_config_requires_it(self, *, no_kafkaesq): + with pytest.raises(ImproperlyConfigured): + ClientConfig(CONFLUENT_CONFIG) + + +@needs_kafkaesq +class TestClientConfig: + def test_from_confluent_keeps_the_config(self): + config = ClientConfig.from_confluent(CONFLUENT_CONFIG) + assert config.as_confluent() == CONFLUENT_CONFIG + # ... as a copy: mutating the result cannot corrupt the source. + config.as_confluent()["group.id"] = "other" + assert config.config["group.id"] == "billing" + + def test_from_aiokafka(self): + config = ClientConfig.from_aiokafka( + bootstrap_servers="localhost:9092", + group_id="billing", + enable_auto_commit=False, + ) + assert config.as_confluent() == { + "bootstrap.servers": "localhost:9092", + "group.id": "billing", + "enable.auto.commit": False, + } + + def test_from_aiokafka_takes_a_mapping(self): + assert ClientConfig.from_aiokafka( + {"bootstrap_servers": ["h1:9092", "h2:9092"]} + ) == ClientConfig.from_confluent({"bootstrap.servers": "h1:9092,h2:9092"}) + + def test_from_aiokafka_reports_unmapped_kwargs(self): + with pytest.raises(KeyError, match="ssl_context"): + ClientConfig.from_aiokafka( + ssl_context=ssl.create_default_context(), on_unmapped="raise" + ) + + def test_as_aiokafka(self): + kwargs = ClientConfig.from_confluent(CONFLUENT_CONFIG).as_aiokafka() + assert kwargs == { + "bootstrap_servers": "h1:9092,h2:9092", + "group_id": "billing", + "client_id": "worker-1", + "session_timeout_ms": 45000, + "auto_offset_reset": "latest", + } + + def test_as_aiokafka_builds_an_ssl_context(self): + kwargs = ClientConfig.from_confluent( + {"bootstrap.servers": "h:9092", "security.protocol": "SSL"} + ).as_aiokafka() + assert isinstance(kwargs["ssl_context"], ssl.SSLContext) + + def test_as_aiokafka_reports_unmapped_keys(self): + with pytest.raises(KeyError, match="socket.keepalive.enable"): + ClientConfig.from_confluent( + {**CONFLUENT_CONFIG, "socket.keepalive.enable": True} + ).as_aiokafka(on_unmapped="raise") + + def test_as_app_settings(self): + assert ClientConfig.from_confluent(CONFLUENT_CONFIG).as_app_settings() == { + "broker": "kafka://h1:9092;kafka://h2:9092", + "id": "billing", + "broker_client_id": "worker-1", + "broker_session_timeout": 45, + "consumer_auto_offset_reset": "latest", + } + + def test_as_app_settings_configures_an_app(self): + settings = ClientConfig.from_confluent(CONFLUENT_CONFIG).as_app_settings() + app = faust.App(settings.pop("id"), **settings) + assert app.conf.id == "billing" + assert [str(url) for url in app.conf.broker] == [ + "kafka://h1:9092", + "kafka://h2:9092", + ] + assert app.conf.broker_session_timeout == 45 + assert app.conf.consumer_auto_offset_reset == "latest" + + def test_as_app_settings_reports_authentication(self): + # Faust takes a broker_credentials object built at runtime, so + # security keys have no setting to convert to. + with pytest.warns(UserWarning, match="sasl.username"): + settings = ClientConfig.from_confluent( + { + **CONFLUENT_CONFIG, + "security.protocol": "sasl_ssl", + "sasl.username": "user", + "sasl.password": "secret", + } + ).as_app_settings() + assert "broker_credentials" not in settings + + def test_round_trips_through_every_spelling(self): + config = ClientConfig.from_confluent(CONFLUENT_CONFIG) + assert ClientConfig.from_aiokafka(config.as_aiokafka()) == config + settings = config.as_app_settings() + assert ClientConfig.from_app_settings( + settings, keys=settings, on_unmapped="raise" + ) == ClientConfig.from_confluent( + # 'broker' converts back with the scheme stripped. + {**CONFLUENT_CONFIG, "bootstrap.servers": "h1:9092,h2:9092"} + ) + + +@needs_kafkaesq +class TestClientConfigFromApp: + def test_consumer_settings(self, *, app): + config = ClientConfig.from_app_settings( + app.conf, CONSUMER_SETTINGS + ).as_confluent() + assert config["group.id"] == app.conf.id + assert config["client.id"] == app.conf.broker_client_id + assert config["session.timeout.ms"] == int( + app.conf.broker_session_timeout * 1000.0 + ) + assert config["heartbeat.interval.ms"] == int( + app.conf.broker_heartbeat_interval * 1000.0 + ) + assert config["max.poll.interval.ms"] == int( + app.conf.broker_max_poll_interval * 1000.0 + ) + assert config["auto.offset.reset"] == app.conf.consumer_auto_offset_reset + assert config["max.partition.fetch.bytes"] == app.conf.consumer_max_fetch_size + assert config["check.crcs"] == app.conf.broker_check_crcs + # Which brokers to talk to comes from the transport, not the settings. + assert "bootstrap.servers" not in config + + def test_producer_settings(self, *, app): + config = ClientConfig.from_app_settings( + app.conf, PRODUCER_SETTINGS + ).as_confluent() + assert config["client.id"] == app.conf.broker_client_id + assert config["acks"] == "all" + assert config["message.max.bytes"] == app.conf.producer_max_request_size + assert "group.id" not in config + + @pytest.mark.conf(broker_session_timeout=45, consumer_auto_offset_reset="latest") + def test_reflects_configured_values(self, *, app): + config = ClientConfig.from_app_settings( + app.conf, CONSUMER_SETTINGS + ).as_confluent() + assert config["session.timeout.ms"] == 45000 + assert config["auto.offset.reset"] == "latest" + + def test_accepts_a_plain_mapping(self): + config = ClientConfig.from_app_settings( + {"id": "billing", "broker_session_timeout": 45}, CONSUMER_SETTINGS + ) + assert config.as_confluent() == { + "group.id": "billing", + "session.timeout.ms": 45000, + } + + def test_unset_settings_are_skipped(self, *, app): + assert app.conf.consumer_group_instance_id is None + config = ClientConfig.from_app_settings(app.conf, CONSUMER_SETTINGS) + assert "group.instance.id" not in config.as_confluent() + + @pytest.mark.conf(consumer_group_instance_id="worker-1") + def test_group_instance_id(self, *, app): + config = ClientConfig.from_app_settings(app.conf, CONSUMER_SETTINGS) + assert config.as_confluent()["group.instance.id"] == "worker-1" + + def test_unmapped_settings_can_be_reported(self, *, app): + with pytest.raises(KeyError, match="table_standby_replicas"): + ClientConfig.from_app_settings( + app.conf, ["id", "table_standby_replicas"], on_unmapped="raise" + ) + + def test_from_app_takes_the_brokers_from_the_transport(self, *, app): + config = ClientConfig.from_app(app, CONSUMER_SETTINGS).as_confluent() + assert config["bootstrap.servers"] == "localhost:9092" + assert config["group.id"] == app.conf.id + + @pytest.mark.conf(broker="kafka://h1:9092;kafka://h2") + def test_from_app_with_several_brokers(self, *, app): + config = ClientConfig.from_app(app).as_confluent() + assert config["bootstrap.servers"] == "h1:9092,h2:9092" + + +@needs_kafkaesq +class TestClientConfigLimits: + def test_producer_timeout_is_clamped(self, *, app): + # producer_request_timeout defaults to 20 minutes; librdkafka refuses + # to build a client above 15, so this would be fatal unclamped. + assert app.conf.producer_request_timeout * 1000.0 > 900000 + config = ClientConfig.from_app_settings(app.conf, PRODUCER_SETTINGS) + assert config.as_confluent()["request.timeout.ms"] == 900000 + + def test_clamping_is_logged(self): + config = ClientConfig.from_confluent({"request.timeout.ms": 1200000}) + with patch.object(mod, "logger") as logger: + assert config.as_confluent() == {"request.timeout.ms": 900000} + logger.warning.assert_called_once() + + def test_value_below_minimum(self): + config = ClientConfig.from_confluent({"message.max.bytes": 10}) + assert config.as_confluent()["message.max.bytes"] == 1000 + + def test_clamping_can_be_turned_off(self): + config = ClientConfig.from_confluent({"request.timeout.ms": 1200000}) + assert config.as_confluent(clamp=False) == {"request.timeout.ms": 1200000} + + def test_values_in_range_are_untouched(self): + source = {"request.timeout.ms": 30000, "linger.ms": 0, "check.crcs": True} + assert ClientConfig.from_confluent(source).as_confluent() == source + + def test_keys_without_a_limit_are_untouched(self): + source = {"batch.num.messages": 10**12} + assert ClientConfig.from_confluent(source).as_confluent() == source + + def test_non_integer_values_are_untouched(self): + source = {"request.timeout.ms": "1200000", "check.crcs": True} + assert ClientConfig.from_confluent(source).as_confluent() == source From d80875d6685f6aafef2ba7e380d7c4d30d320e3c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 22:24:41 +0000 Subject: [PATCH 2/2] Document the confluent driver in the API reference faust.transport.drivers.aiokafka has a reference page; its confluent counterpart never did, and was silenced in conf.py's apicheck_ignore_modules so `make apicheck` would not report the gap. Give it a page, drop the silencer, and list both drivers together under Transports. The docs build imports every documented module for autodoc and the confluent driver imports confluent_kafka at module level (unlike the optional stores, which guard theirs), so the docs environment now installs the ckafka extra. The broker setting's transport list gains a reference link per driver. That list lives in the setting's docstring, which docs/includes/settingref.txt is generated from, so the client-config-conversion section added there earlier moves into the docstring too -- the generated file now matches what `make configref` produces for that section, instead of drifting from it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01BYhhm6H1FQskUmDqjrft7G --- docs/conf.py | 1 - docs/includes/settingref.txt | 40 +++++++------ .../faust.transport.drivers.confluent.rst | 11 ++++ docs/reference/index.rst | 1 + faust/transport/kafkaesq.py | 4 +- faust/types/settings/settings.py | 59 +++++++++++++++++++ requirements/docs.txt | 3 + 7 files changed, 99 insertions(+), 20 deletions(-) create mode 100644 docs/reference/faust.transport.drivers.confluent.rst diff --git a/docs/conf.py b/docs/conf.py index 5f2a03ed5..71cffde0e 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -62,7 +62,6 @@ 'faust.cli', 'faust.models', 'faust.serializers', - 'faust.transport.drivers.confluent', 'faust.types', 'faust.types._env', 'faust.utils', diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 65ff3fe82..ad15ebf06 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -390,6 +390,8 @@ You can also pass a list of URLs: Limitations: None + Reference: :mod:`faust.transport.drivers.aiokafka` + - ``confluent://`` @@ -399,23 +401,25 @@ You can also pass a list of URLs: suitable for tables), and do not create any necessary internal topics (you have to create them manually). + Reference: :mod:`faust.transport.drivers.confluent` + .. _client-config-conversion: **Converting client configuration** -The clients behind these transports spell their configuration differently: -:pypi:`aiokafka` takes ``session_timeout_ms``, librdkafka takes -``session.timeout.ms``, and app settings are a third spelling again -(:setting:`broker_session_timeout`, in seconds). Install the ``kafkaesq`` -bundle: +The clients behind these transports spell their configuration +differently: :pypi:`aiokafka` takes ``session_timeout_ms``, librdkafka +takes ``session.timeout.ms``, and app settings are a third spelling +again (:setting:`broker_session_timeout`, in seconds). Install the +``kafkaesq`` bundle: .. sourcecode:: console $ pip install "faust-streaming[kafkaesq]" -and :class:`faust.transport.kafkaesq.ClientConfig` converts between them. A -librdkafka config -- a Confluent Cloud ``client.properties``, say -- becomes -arguments for :class:`faust.App`: +and :class:`faust.transport.kafkaesq.ClientConfig` converts between +them. A librdkafka config -- a Confluent Cloud ``client.properties``, +say -- becomes arguments for :class:`faust.App`: .. sourcecode:: python @@ -430,8 +434,8 @@ arguments for :class:`faust.App`: settings = config.as_app_settings() app = faust.App(settings.pop('id'), **settings) -and, the other way round, an app's settings configure a plain client that -talks to the same brokers: +and, the other way round, an app's settings configure a plain client +that talks to the same brokers: .. sourcecode:: python @@ -442,17 +446,17 @@ talks to the same brokers: # ... or the same settings as aiokafka kwargs producer = AIOKafkaProducer(**config.as_aiokafka()) -This is a conversion helper, not something the transports use: both configure -themselves from your app settings as they always have, and neither needs -:pypi:`kafkaesq` installed. +This is a conversion helper, not something the transports use: both +configure themselves from your app settings as they always have, and +neither needs :pypi:`kafkaesq` installed. .. note:: - Authentication is not converted. Faust configures it with a - :setting:`broker_credentials` object built at runtime, which no config - file can describe, so ``security.protocol``, ``sasl.*`` and ``ssl.*`` - keys are reported as unmapped when converting to app settings -- set - :setting:`broker_credentials` yourself. + Authentication is not converted. Faust configures it with a + :setting:`broker_credentials` object built at runtime, which no + config file can describe, so ``security.protocol``, ``sasl.*`` + and ``ssl.*`` keys are reported as unmapped when converting to + app settings -- set :setting:`broker_credentials` yourself. .. setting:: broker_credentials diff --git a/docs/reference/faust.transport.drivers.confluent.rst b/docs/reference/faust.transport.drivers.confluent.rst new file mode 100644 index 000000000..8a21b8004 --- /dev/null +++ b/docs/reference/faust.transport.drivers.confluent.rst @@ -0,0 +1,11 @@ +===================================================== + ``faust.transport.drivers.confluent`` +===================================================== + +.. contents:: + :local: +.. currentmodule:: faust.transport.drivers.confluent + +.. automodule:: faust.transport.drivers.confluent + :members: + :undoc-members: diff --git a/docs/reference/index.rst b/docs/reference/index.rst index 9bdb3a2e6..3f632ecf7 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -159,6 +159,7 @@ Transports faust.transport.producer faust.transport.drivers faust.transport.drivers.aiokafka + faust.transport.drivers.confluent faust.transport.kafkaesq faust.transport.utils diff --git a/faust/transport/kafkaesq.py b/faust/transport/kafkaesq.py index a03df33bc..49130a577 100644 --- a/faust/transport/kafkaesq.py +++ b/faust/transport/kafkaesq.py @@ -25,7 +25,9 @@ app = faust.App(settings.pop('id'), **settings) The conversion tables are :pypi:`kafkaesq`'s, installed with the ``kafkaesq`` -bundle:: +bundle: + +.. sourcecode:: console $ pip install "faust-streaming[kafkaesq]" diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index cf080dab8..8099cc772 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -705,6 +705,8 @@ def broker(self) -> List[URL]: Limitations: None + Reference: :mod:`faust.transport.drivers.aiokafka` + - ``confluent://`` @@ -713,6 +715,63 @@ def broker(self) -> List[URL]: Limitations: Does not do sticky partition assignment (not suitable for tables), and do not create any necessary internal topics (you have to create them manually). + + Reference: :mod:`faust.transport.drivers.confluent` + + .. _client-config-conversion: + + **Converting client configuration** + + The clients behind these transports spell their configuration + differently: :pypi:`aiokafka` takes ``session_timeout_ms``, librdkafka + takes ``session.timeout.ms``, and app settings are a third spelling + again (:setting:`broker_session_timeout`, in seconds). Install the + ``kafkaesq`` bundle: + + .. sourcecode:: console + + $ pip install "faust-streaming[kafkaesq]" + + and :class:`faust.transport.kafkaesq.ClientConfig` converts between + them. A librdkafka config -- a Confluent Cloud ``client.properties``, + say -- becomes arguments for :class:`faust.App`: + + .. sourcecode:: python + + from faust.transport.kafkaesq import ClientConfig + + config = ClientConfig.from_confluent({ + 'bootstrap.servers': 'kafka.example.com:9092', + 'group.id': 'billing', + 'session.timeout.ms': 45000, + }) + + settings = config.as_app_settings() + app = faust.App(settings.pop('id'), **settings) + + and, the other way round, an app's settings configure a plain client + that talks to the same brokers: + + .. sourcecode:: python + + from faust.transport.kafkaesq import PRODUCER_SETTINGS, ClientConfig + + config = ClientConfig.from_app(app, PRODUCER_SETTINGS) + producer = confluent_kafka.Producer(config.as_confluent()) + # ... or the same settings as aiokafka kwargs + producer = AIOKafkaProducer(**config.as_aiokafka()) + + This is a conversion helper, not something the transports use: both + configure themselves from your app settings as they always have, and + neither needs :pypi:`kafkaesq` installed. + + .. note:: + + Authentication is not converted. Faust configures it with a + :setting:`broker_credentials` object built at runtime, which no + config file can describe, so ``security.protocol``, ``sasl.*`` + and ``ssl.*`` keys are reported as unmapped when converting to + app settings -- set :setting:`broker_credentials` yourself. """ @broker.on_set_default # type: ignore diff --git a/requirements/docs.txt b/requirements/docs.txt index 0edd8a038..2f4e598d3 100644 --- a/requirements/docs.txt +++ b/requirements/docs.txt @@ -1,4 +1,7 @@ -r requirements.txt +# autodoc imports every documented module, and the confluent driver +# imports confluent_kafka at module level. +-r extras/ckafka.txt sphinx<6.0.0 sphinx-celery sphinx-autodoc-typehints