From 334d0462c198c0dfd4f25e09d5fda3ab330e8b12 Mon Sep 17 00:00:00 2001 From: Valerij Maljulin Date: Wed, 9 Sep 2026 16:33:19 +0200 Subject: [PATCH 1/2] Add Kafka messaging support JIRA: RHELWF-13775 Assisted-by: Opus 4.6 --- conf/settings.py.example | 21 ++++- pyproject.toml | 2 + resultsdb/__init__.py | 5 +- resultsdb/config.py | 17 ++++- resultsdb/messaging.py | 80 ++++++++++++++++++- tests/test_app.py | 40 +++++++++- tests/test_general.py | 14 ++++ tests/test_messaging_kafka.py | 139 ++++++++++++++++++++++++++++++++++ uv.lock | 25 ++++++ 9 files changed, 336 insertions(+), 7 deletions(-) create mode 100644 tests/test_messaging_kafka.py diff --git a/conf/settings.py.example b/conf/settings.py.example index a409f3f..71564b9 100644 --- a/conf/settings.py.example +++ b/conf/settings.py.example @@ -71,11 +71,11 @@ OIDC_USERNAME_FIELD = 'uid' MESSAGE_BUS_PUBLISH = False # Name of the message bus plugin to use goes here. 'fedmsg' is installed by # default, but you could create your own. -# Supported values: 'dummy', 'stomp', 'fedmsg' +# Supported values: 'dummy', 'stomp', 'fedmsg', 'kafka' MESSAGE_BUS_PLUGIN = 'fedmsg' MESSAGE_BUS_KWARGS = {'modname': 'resultsdb'} -## Alternatively, you could use the 'stomp' messaging plugin. +## Alternatively, you could use the 'stomp' messaging plugin (UMB). #MESSAGE_BUS_PLUGIN = 'stomp' #MESSAGE_BUS_KWARGS = { # 'destination': '/topic/VirtualTopic.eng.resultsdb.result.new', @@ -91,5 +91,22 @@ MESSAGE_BUS_KWARGS = {'modname': 'resultsdb'} # }, #} +## Alternatively, you could use the 'kafka' messaging plugin. +## SASL credentials come from RESULTSDB_KAFKA_SASL_USERNAME and +## RESULTSDB_KAFKA_SASL_PASSWORD environment variables. +#MESSAGE_BUS_PLUGIN = 'kafka' +#KAFKA = { +# 'topic': 'eng.resultsdb.result.new', +# 'producer': { +# 'security.protocol': 'SASL_SSL', +# 'sasl.mechanisms': 'SCRAM-SHA-512', +# 'bootstrap.servers': 'broker01:9096,broker02:9096', +# 'client.id': 'resultsdb', +# 'retries': 10, +# 'retry.backoff.ms': 100, +# }, +# 'flush_timeout_seconds': 20.0, +#} + # Publish Taskotron-compatible fedmsgs on the 'taskotron' topic MESSAGE_BUS_PUBLISH_TASKOTRON = False diff --git a/pyproject.toml b/pyproject.toml index 21dbb3c..f8ce1ac 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "gunicorn>=26.2.0,<26.3.0", "fedora-messaging>=3.4.1,<4.0.0", "stomp-py>=9.0.0,<9.1.0", + "confluent-kafka>=2.14.0,<3.0.0", "Flask-SQLAlchemy>=3.1.1,<4.0.0", "SQLAlchemy>=2.0.36,<3.0.0", "psycopg2-binary>=2.9.10,<3.0.0", @@ -44,6 +45,7 @@ resultsdb = "resultsdb.__main__:cli" dummy = "resultsdb.messaging:DummyPlugin" fedmsg = "resultsdb.messaging:FedmsgPlugin" stomp = "resultsdb.messaging:StompPlugin" +kafka = "resultsdb.messaging:KafkaPlugin" [project.urls] Homepage = "https://docs.resultsdb20.apiary.io/" diff --git a/resultsdb/__init__.py b/resultsdb/__init__.py index 8f1a7e7..252dfd4 100644 --- a/resultsdb/__init__.py +++ b/resultsdb/__init__.py @@ -188,7 +188,10 @@ def setup_messaging(app): plugin_name = app.config["MESSAGE_BUS_PLUGIN"] app.logger.info("Using messaging plugin %s", plugin_name) - plugin_args = app.config["MESSAGE_BUS_KWARGS"] + if plugin_name == "kafka": + plugin_args = app.config.get("KAFKA") or {} + else: + plugin_args = app.config["MESSAGE_BUS_KWARGS"] app.messaging_plugin = load_messaging_plugin( name=plugin_name, plugin_args=plugin_args, diff --git a/resultsdb/config.py b/resultsdb/config.py index 0bd558b..b5c0d22 100644 --- a/resultsdb/config.py +++ b/resultsdb/config.py @@ -108,7 +108,7 @@ class Config: MESSAGE_BUS_PUBLISH = True # Name of the message bus plugin to use goes here. 'fedmsg' is installed by # default, but you could create your own. - # Supported values: 'dummy', 'stomp', 'fedmsg' + # Supported values: 'dummy', 'stomp', 'fedmsg', 'kafka' MESSAGE_BUS_PLUGIN = "dummy" # You can pass extra arguments to your message bus plugin here. For instance, # the fedmsg plugin expects an extra `modname` argument that can be used to @@ -116,6 +116,21 @@ class Config: # ... # e.g. org.fedoraproject.prod.resultsdb.result.new MESSAGE_BUS_KWARGS: dict[str, object] = {} + # Used when MESSAGE_BUS_PLUGIN is 'kafka'. + # 'producer' keys are passed directly to confluent-kafka (librdkafka). + # Full reference: https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md + # SASL credentials are read from RESULTSDB_KAFKA_SASL_USERNAME and + # RESULTSDB_KAFKA_SASL_PASSWORD environment variables. + KAFKA: dict[str, object] = { + "topic": "eng.resultsdb.result.new", + "producer": { + "bootstrap.servers": "localhost:9092", + "client.id": "resultsdb", + "retries": 3, + "retry.backoff.ms": 100, + }, + "flush_timeout_seconds": 20.0, + } # Publish Taskotron-compatible fedmsgs on the 'taskotron' topic MESSAGE_BUS_PUBLISH_TASKOTRON = False diff --git a/resultsdb/messaging.py b/resultsdb/messaging.py index c6a589f..04450df 100644 --- a/resultsdb/messaging.py +++ b/resultsdb/messaging.py @@ -20,11 +20,14 @@ import abc import json import logging +import os from importlib.metadata import entry_points from threading import Lock -from typing import ClassVar +from typing import Any, ClassVar import stomp +from confluent_kafka import KafkaError, KafkaException, Producer +from confluent_kafka import Message as KafkaMessage from fedora_messaging.api import Message, publish from fedora_messaging.exceptions import ( ConnectionException, @@ -36,6 +39,7 @@ from opentelemetry.trace.propagation.tracecontext import ( TraceContextTextMapPropagator, ) +from pydantic import BaseModel from tenacity import retry, stop_after_attempt, wait_exponential from resultsdb.models import db @@ -252,10 +256,82 @@ def _publish_with_retry(self, **kwargs): log.debug("Published message through stomp: %s", kwargs["body"]) +class KafkaConfig(BaseModel): + topic: str + producer: dict[str, Any] + flush_timeout_seconds: float = 20.0 + + +def _parse_kafka_config(config: dict[str, Any]) -> tuple[KafkaConfig, dict[str, Any]]: + try: + kafka_config = KafkaConfig(**config) + except (ValueError, TypeError) as e: + raise RuntimeError(f"Invalid KAFKA configuration: {e}") from e + + username = os.environ.get("RESULTSDB_KAFKA_SASL_USERNAME") + password = os.environ.get("RESULTSDB_KAFKA_SASL_PASSWORD") + if not username or not password: + raise RuntimeError( + "RESULTSDB_KAFKA_SASL_USERNAME and RESULTSDB_KAFKA_SASL_PASSWORD " + "environment variables are required" + ) + producer_config = { + **kafka_config.producer, + "sasl.username": username, + "sasl.password": password, + } + + return kafka_config, producer_config + + +class KafkaPlugin(MessagingPlugin): + """A Kafka plugin used to publish to a Kafka (MSK) cluster.""" + + def __init__(self, **kwargs): + kafka_config, producer_config = _parse_kafka_config(kwargs) + self._config = kafka_config + self._producer = Producer(producer_config) + + @tracer.start_as_current_span("KafkaPlugin.publish") + def publish(self, message): + # Add telemetry information. This includes an extra key + # traceparent. + TraceContextTextMapPropagator().inject(message) + + delivery_error = None + + def _delivery_callback(err: KafkaError | None, _msg: KafkaMessage) -> None: + nonlocal delivery_error + if err is not None: + log.error("Failed to deliver Kafka message: %s", err) + delivery_error = delivery_error or KafkaException(err) + else: + log.debug("Published message through kafka: %s", message) + + self._producer.produce( + self._config.topic, + value=json.dumps(message).encode("utf-8"), + on_delivery=_delivery_callback, + ) + + remaining = self._producer.flush(timeout=self._config.flush_timeout_seconds) + if remaining > 0: + log.error("%d Kafka message(s) failed to be delivered (timeout)", remaining) + raise KafkaException( + KafkaError( + KafkaError._MSG_TIMED_OUT, + f"{remaining} message(s) were not delivered within timeout", + ) + ) + + if delivery_error is not None: + raise delivery_error + + def load_messaging_plugin(name, plugin_args): """Instantiate and return the appropriate messaging plugin.""" points = entry_points(group="resultsdb.messaging.plugins") - classes = {"dummy": DummyPlugin} + classes = {"dummy": DummyPlugin, "kafka": KafkaPlugin} classes.update({point.name: point.load() for point in points}) log.debug("Found the following installed messaging plugin %r", classes) diff --git a/tests/test_app.py b/tests/test_app.py index 3197f7a..1026e8a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,4 +1,4 @@ -from unittest.mock import Mock +from unittest.mock import Mock, patch from pytest import raises from werkzeug.test import EnvironBuilder @@ -61,3 +61,41 @@ def test_app_messaging_stomp_bad(): expected_error = "Missing 'destination' option for STOMP messaging plugin" with raises(ValueError, match=expected_error): setup_messaging(app) + + +def test_app_messaging_kafka(monkeypatch): + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_USERNAME", "alice") + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_PASSWORD", "secret") + app = Mock() + app.config = { + "MESSAGE_BUS_PUBLISH": True, + "MESSAGE_BUS_PLUGIN": "kafka", + "KAFKA": { + "topic": "eng.resultsdb.result.new", + "producer": { + "bootstrap.servers": "localhost:9092", + }, + }, + "MESSAGE_BUS_KWARGS": {}, + } + with patch("resultsdb.messaging.Producer"): + setup_messaging(app) + app.logger.info.assert_called_once_with("Using messaging plugin %s", "kafka") + assert type(app.messaging_plugin).__name__ == "KafkaPlugin" + + +def test_app_messaging_kafka_bad(monkeypatch): + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_USERNAME", "alice") + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_PASSWORD", "secret") + app = Mock() + app.config = { + "MESSAGE_BUS_PUBLISH": True, + "MESSAGE_BUS_PLUGIN": "kafka", + "KAFKA": { + "producer": { + "bootstrap.servers": "localhost:9092", + }, + }, + } + with raises(RuntimeError, match="Invalid KAFKA configuration"): + setup_messaging(app) diff --git a/tests/test_general.py b/tests/test_general.py index 4f3d7fa..b85603a 100644 --- a/tests/test_general.py +++ b/tests/test_general.py @@ -175,6 +175,20 @@ def test_stomp_load(self): assert isinstance(plugin, messaging.StompPlugin) assert plugin.destination == "results.new" + def test_kafka_load(self, monkeypatch): + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_USERNAME", "alice") + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_PASSWORD", "secret") + kafka_kwargs = { + "topic": "eng.resultsdb.result.new", + "producer": { + "bootstrap.servers": "localhost:9092", + }, + } + with patch("resultsdb.messaging.Producer"): + plugin = messaging.load_messaging_plugin("kafka", kafka_kwargs) + assert isinstance(plugin, messaging.KafkaPlugin) + assert plugin._config.topic == "eng.resultsdb.result.new" + def test_stomp_missing_destination(self): message_bus_kwargs = { "connection": { diff --git a/tests/test_messaging_kafka.py b/tests/test_messaging_kafka.py new file mode 100644 index 0000000..d499f07 --- /dev/null +++ b/tests/test_messaging_kafka.py @@ -0,0 +1,139 @@ +import json +from unittest.mock import Mock, patch + +from confluent_kafka import KafkaError, KafkaException +from pytest import fixture, raises + +from resultsdb.messaging import KafkaPlugin + + +@fixture +def kafka_plugin_args(monkeypatch): + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_USERNAME", "alice") + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_PASSWORD", "secret") + return { + "topic": "eng.resultsdb.result.new", + "producer": { + "bootstrap.servers": "localhost:9092", + "client.id": "resultsdb-test", + "retries": 3, + }, + "flush_timeout_seconds": 15.0, + } + + +@fixture +def kafka_plugin(kafka_plugin_args): + with patch("resultsdb.messaging.Producer") as mock_producer_class: + mock_producer = Mock() + mock_producer.flush.return_value = 0 + mock_producer_class.return_value = mock_producer + plugin = KafkaPlugin(**kafka_plugin_args) + yield plugin, mock_producer + + +def _simulate_successful_produce(mock_producer): + """Make produce() invoke the on_delivery callback with no error on flush().""" + callbacks = [] + + def capture_produce(topic, value=None, on_delivery=None): + callbacks.append(on_delivery) + + def trigger_flush(timeout=None): + for cb in callbacks: + if cb: + cb(None, Mock()) + callbacks.clear() + return 0 + + mock_producer.produce.side_effect = capture_produce + mock_producer.flush.side_effect = trigger_flush + + +def _simulate_failed_produce(mock_producer, error): + """Make produce() invoke the on_delivery callback with an error on flush().""" + callbacks = [] + + def capture_produce(topic, value=None, on_delivery=None): + callbacks.append(on_delivery) + + def trigger_flush(timeout=None): + for cb in callbacks: + if cb: + cb(error, None) + callbacks.clear() + return 0 + + mock_producer.produce.side_effect = capture_produce + mock_producer.flush.side_effect = trigger_flush + + +def test_publish_success(kafka_plugin): + plugin, mock_producer = kafka_plugin + _simulate_successful_produce(mock_producer) + + plugin.publish({"id": 1, "outcome": "PASSED"}) + + mock_producer.produce.assert_called_once() + args, kwargs = mock_producer.produce.call_args + assert args[0] == "eng.resultsdb.result.new" + payload = json.loads(kwargs["value"]) + assert payload["id"] == 1 + assert payload["outcome"] == "PASSED" + mock_producer.flush.assert_called_once_with(timeout=15.0) + + +def test_publish_delivery_error(kafka_plugin): + plugin, mock_producer = kafka_plugin + mock_error = Mock() + mock_error.__str__ = Mock(return_value="Connection failed") + _simulate_failed_produce(mock_producer, mock_error) + + with raises(KafkaException): + plugin.publish({"id": 1}) + + +def test_publish_flush_timeout(kafka_plugin): + plugin, mock_producer = kafka_plugin + mock_producer.flush.return_value = 1 + + with raises(KafkaException) as exc_info: + plugin.publish({"id": 1}) + + err = exc_info.value.args[0] + assert isinstance(err, KafkaError) + assert err.code() == KafkaError._MSG_TIMED_OUT + + +def test_kafka_plugin_creation(kafka_plugin_args): + with patch("resultsdb.messaging.Producer") as mock_producer_class: + mock_producer = Mock() + mock_producer_class.return_value = mock_producer + + plugin = KafkaPlugin(**kafka_plugin_args) + + mock_producer_class.assert_called_once_with( + { + "bootstrap.servers": "localhost:9092", + "client.id": "resultsdb-test", + "retries": 3, + "sasl.username": "alice", + "sasl.password": "secret", + } + ) + assert plugin._config.topic == "eng.resultsdb.result.new" + assert plugin._producer is mock_producer + + +def test_kafka_plugin_missing_sasl(kafka_plugin_args, monkeypatch): + monkeypatch.delenv("RESULTSDB_KAFKA_SASL_USERNAME", raising=False) + monkeypatch.delenv("RESULTSDB_KAFKA_SASL_PASSWORD", raising=False) + with raises(RuntimeError, match="RESULTSDB_KAFKA_SASL_USERNAME"): + KafkaPlugin(**kafka_plugin_args) + + +def test_kafka_plugin_invalid_config(monkeypatch): + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_USERNAME", "alice") + monkeypatch.setenv("RESULTSDB_KAFKA_SASL_PASSWORD", "secret") + with raises(RuntimeError, match="Invalid KAFKA configuration"): + KafkaPlugin(producer={"bootstrap.servers": "localhost:9092"}) diff --git a/uv.lock b/uv.lock index 7a1444b..e027cc8 100644 --- a/uv.lock +++ b/uv.lock @@ -253,6 +253,29 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "confluent-kafka" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/51/90/eb998fedefb63b42910b54b76b7d300ccddc56430a5175122eb60cedc4f3/confluent_kafka-2.15.0.tar.gz", hash = "sha256:7ad9bad1cbabf6713ec039b8204b48d322024fd11397eec88d912e048c732ba7", size = 323365, upload-time = "2026-06-30T19:48:43.395Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/7a/b3/34811eb66633dce51f583cd0ea5d78a26c9a135847d7dbd536fc5a963055/confluent_kafka-2.15.0-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:6e5ea2933c0b07782a99dfdfc9d52e92526e7a61839a6994feb994dcf2a78a74", size = 4367689, upload-time = "2026-06-30T19:48:00.855Z" }, + { url = "https://files.pythonhosted.org/packages/8f/08/9fea6928cd1e678df0be930206e631bb7e44e7ed54560f5b3f76352187ff/confluent_kafka-2.15.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:e4465a060ab215d5d514b181f3e7502c6fd7887529381c4af940a0b60f9d65cc", size = 4338594, upload-time = "2026-06-30T19:48:02.469Z" }, + { url = "https://files.pythonhosted.org/packages/25/5f/f9728a100ba40d31a957c5f5cb594ea7d49fd1188002543d3543349a063a/confluent_kafka-2.15.0-cp312-cp312-manylinux_2_28_aarch64.whl", hash = "sha256:b68e6f441866542c9150e6fdf106131eee5a51db591937888c866d799fb354ec", size = 4979384, upload-time = "2026-06-30T19:48:04.127Z" }, + { url = "https://files.pythonhosted.org/packages/8b/2e/9f80cae9dd7b4852b8a24f9f91001916b987a2ba359e9f077c43ca41d5da/confluent_kafka-2.15.0-cp312-cp312-manylinux_2_28_x86_64.whl", hash = "sha256:fd833dab1392f456b9dd52b7ed9ac65b2cc36871ed5e3ebe8d53486c965b6453", size = 4785580, upload-time = "2026-06-30T19:48:05.841Z" }, + { url = "https://files.pythonhosted.org/packages/6e/b6/02f73ed259d5be5fbbc8dde41527cad097eca56ab7831026039c8abbaffc/confluent_kafka-2.15.0-cp312-cp312-win_amd64.whl", hash = "sha256:2cc8d01a77c534669bb896f731c8917cd63b787618081fdd0d1420d58cd6815b", size = 4549188, upload-time = "2026-06-30T19:48:07.426Z" }, + { url = "https://files.pythonhosted.org/packages/b0/80/c7bba18644fb748fc1202e2652230652d243c764fc158798d9f2096513c4/confluent_kafka-2.15.0-cp313-cp313-macosx_13_0_arm64.whl", hash = "sha256:ff43508f8e929d83545272ef4f17e27fdaa9cc8397f77cf2236b22d67de4edc4", size = 4343586, upload-time = "2026-06-30T19:48:09.272Z" }, + { url = "https://files.pythonhosted.org/packages/fa/41/0be8143a944c70c6055e8d0fb713964ebe173494e280451c927fa02247ab/confluent_kafka-2.15.0-cp313-cp313-macosx_13_0_x86_64.whl", hash = "sha256:41d4c7360c51202785e8e9aa56241dd392b9dbcc311eae7c5e6985b239efc12b", size = 4371325, upload-time = "2026-06-30T19:48:10.996Z" }, + { url = "https://files.pythonhosted.org/packages/e2/3c/d5c83c20599a82faa065b922b55338c672a268fa2ac75f54e53bed8913d7/confluent_kafka-2.15.0-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:b0ac3831aac47ea23699a6ca5b9836288a4b9d201015dbfc4c3ec9f20592fdb5", size = 4979865, upload-time = "2026-06-30T19:48:12.447Z" }, + { url = "https://files.pythonhosted.org/packages/40/79/5be34236697e8fabbed5534a67100b9d9d11c210de55ba3ba2590d7d7634/confluent_kafka-2.15.0-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:0455754e8294e1e76cb5acc083c300b6bd8d88f2d2c551c583cbf4ab55889a57", size = 4785955, upload-time = "2026-06-30T19:48:14.055Z" }, + { url = "https://files.pythonhosted.org/packages/94/ff/4da4e954040769d46111f3c0e5b3e76f2d3bc4b27c1deb5fd64fc9cc4055/confluent_kafka-2.15.0-cp313-cp313-win_amd64.whl", hash = "sha256:7c59ea36606e0e946e6b353203136b4d71be6d95c0c8ae5113b81850ff5b7b41", size = 4608827, upload-time = "2026-06-30T19:48:15.746Z" }, + { url = "https://files.pythonhosted.org/packages/18/20/5c078b9560b8005404e09a5889b9771e8deb5c9dddeb9003f58c7d028258/confluent_kafka-2.15.0-cp314-cp314-macosx_13_0_arm64.whl", hash = "sha256:e781d39ff31d5d10d0502471f92f95aff4f5be1eaae9a1f57aeac164bc9f9029", size = 4343440, upload-time = "2026-06-30T19:48:17.324Z" }, + { url = "https://files.pythonhosted.org/packages/aa/27/fd83281231b6179e3923822e20861dfc1d6c200edb910c3d8e9a15b9e95a/confluent_kafka-2.15.0-cp314-cp314-macosx_13_0_x86_64.whl", hash = "sha256:000463c822a7adc3293369b63688b68d17c03a1a4a64f86182efc08f8b150676", size = 4371051, upload-time = "2026-06-30T19:48:19.29Z" }, + { url = "https://files.pythonhosted.org/packages/5e/3e/fa82e6699144e707972bbf57489598defdf67e844ae2a7d29e0ea4e3a187/confluent_kafka-2.15.0-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:9ddf4cf4647e5d633ef64e3f5a349d5288edfedf974203993e9d86548c2695be", size = 4979624, upload-time = "2026-06-30T19:48:21.454Z" }, + { url = "https://files.pythonhosted.org/packages/13/ee/b4b6a0da17584b432c83a0500ac79c04864ef92dafdb405614831b995232/confluent_kafka-2.15.0-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:d8ed33f623ff2a104fb76f99a67e9917f0170fddb4e28380fcdc83347b1646b2", size = 4785678, upload-time = "2026-06-30T19:48:23.068Z" }, + { url = "https://files.pythonhosted.org/packages/a8/f4/19a852d16e8e4f8ac930037d8ecda21a220f6d5d050a59bbce10189ac9ec/confluent_kafka-2.15.0-cp314-cp314-win_amd64.whl", hash = "sha256:2e80bd96f61aae2ffba951754a769e6d3c5ebb5a5e778ed0ae8ad899ea91556b", size = 4744444, upload-time = "2026-06-30T19:48:24.698Z" }, +] + [[package]] name = "constantly" version = "23.10.4" @@ -1430,6 +1453,7 @@ version = "2.2.0" source = { editable = "." } dependencies = [ { name = "alembic" }, + { name = "confluent-kafka" }, { name = "email-validator" }, { name = "fedora-messaging" }, { name = "flask" }, @@ -1460,6 +1484,7 @@ test = [ [package.metadata] requires-dist = [ { name = "alembic", specifier = ">=1.13.3,<2.0.0" }, + { name = "confluent-kafka", specifier = ">=2.14.0,<3.0.0" }, { name = "email-validator", specifier = ">=2.2.0,<3.0.0" }, { name = "fedora-messaging", specifier = ">=3.4.1,<4.0.0" }, { name = "flask", specifier = ">=3.0.1,<4.0.0" }, From 75772fe27b0a14d7e83ade065221496b9ee40cf1 Mon Sep 17 00:00:00 2001 From: mvalik <42725234+mvalik@users.noreply.github.com> Date: Tue, 15 Sep 2026 10:43:25 +0200 Subject: [PATCH 2/2] Update resultsdb/__init__.py Co-authored-by: Lukas Holecek --- resultsdb/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/resultsdb/__init__.py b/resultsdb/__init__.py index 252dfd4..67e3457 100644 --- a/resultsdb/__init__.py +++ b/resultsdb/__init__.py @@ -189,7 +189,7 @@ def setup_messaging(app): plugin_name = app.config["MESSAGE_BUS_PLUGIN"] app.logger.info("Using messaging plugin %s", plugin_name) if plugin_name == "kafka": - plugin_args = app.config.get("KAFKA") or {} + plugin_args = app.config["KAFKA"] else: plugin_args = app.config["MESSAGE_BUS_KWARGS"] app.messaging_plugin = load_messaging_plugin(