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/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..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,6 +401,63 @@ 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: + +.. 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.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/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..3f632ecf7 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -159,6 +159,8 @@ Transports faust.transport.producer faust.transport.drivers faust.transport.drivers.aiokafka + faust.transport.drivers.confluent + 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..49130a577 --- /dev/null +++ b/faust/transport/kafkaesq.py @@ -0,0 +1,351 @@ +"""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: + +.. sourcecode:: console + + $ 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/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 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