Add Kafka messaging support

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

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/"

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["KAFKA"]
+    else:
+        plugin_args = app.config["MESSAGE_BUS_KWARGS"]
     app.messaging_plugin = load_messaging_plugin(
         name=plugin_name,
         plugin_args=plugin_args,

resultsdb/config.py

@@ -108,14 +108,29 @@ 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
     # configure the topic, like this:
     #   <topic_prefix>.<environment>.<modname>.<topic>
     # 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

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)

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)

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": {