PubSub: add support for Kafka (#2259)

Author: tomkralidis

docs/source/pubsub.rst

@@ -85,6 +85,11 @@ Brokers
 
 The following protocols are supported:
 
+.. note::
+
+   Pub/Sub client dependencies will vary based on the selected broker.  ``requirements-pubsub.txt`` contains all requirements for supported brokers, as a reference point.
+
+
 MQTT
 ^^^^
 
@@ -99,6 +104,23 @@ Example directive:
            channel: messages/a/data  # optional
            hidden: false # default
 
+Kafka
+^^^^^
+
+Example directive:
+
+.. code-block:: yaml
+
+   pubsub:
+       name: Kafka
+       broker:
+           url: tcp://localhost:9092
+           channel: messages-a-data
+           # if using authentication:
+           # sasl_mechanism: PLAIN  # default PLAIN
+           # sasl_security_protocol: SASL_PLAINTEXT  # default SASL_PLAINTEXT
+           hidden: true  # default false
+
 HTTP
 ^^^^
 
@@ -113,12 +135,16 @@ Example directive:
            channel: messages-a-data  # optional
            hidden: true  # default false
 
+Additional information
+----------------------
+
 .. note::
 
    For any Pub/Sub endpoints requiring authentication, encode the ``url`` value as follows:
 
    * ``mqtt://username:password@localhost:1883``
    * ``https://username:password@localhost``
+   * ``tcp://username:password@localhost:9092``
 
    As with any section of the pygeoapi configuration, environment variables may be used as needed, for example
    to set username/password information in a URL.  If ``pubsub.broker.url`` contains authentication, and
@@ -131,5 +157,6 @@ Example directive:
 
    If a ``channel`` is not defined, only the relevant OGC API endpoint is used.
 
+
 .. _`OGC API Publish-Subscribe Workflow - Part 1: Core`: https://docs.ogc.org/DRAFTS/25-030.html
 .. _`AsyncAPI`: https://www.asyncapi.com

pygeoapi/plugin.py

@@ -88,6 +88,7 @@
     },
     'pubsub': {
         'HTTP': 'pygeoapi.pubsub.http.HTTPPubSubClient',
+        'Kafka': 'pygeoapi.pubsub.kafka.KafkaPubSubClient',
         'MQTT': 'pygeoapi.pubsub.mqtt.MQTTPubSubClient'
     }
 }

pygeoapi/pubsub/http.py

@@ -41,7 +41,7 @@
 class HTTPPubSubClient(BasePubSubClient):
     """HTTP client"""
 
-    def __init__(self, broker_url):
+    def __init__(self, publisher_def):
         """
         Initialize object
 
@@ -50,7 +50,7 @@ def __init__(self, broker_url):
         :returns: pygeoapi.pubsub.http.HTTPPubSubClient
         """
 
-        super().__init__(broker_url)
+        super().__init__(publisher_def)
         self.name = 'HTTP'
         self.type = 'http'
         self.auth = None

pygeoapi/pubsub/kafka.py

@@ -0,0 +1,109 @@
+# =================================================================
+#
+# Authors: Tom Kralidis <tomkralidis@gmail.com>
+#
+# Copyright (c) 2026 Tom Kralidis
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation
+# files (the "Software"), to deal in the Software without
+# restriction, including without limitation the rights to use,
+# copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the
+# Software is furnished to do so, subject to the following
+# conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
+# OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
+# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+#
+# =================================================================
+
+import logging
+
+from kafka import errors, KafkaProducer
+
+from pygeoapi.pubsub.base import BasePubSubClient, PubSubClientConnectionError
+from pygeoapi.util import to_json
+
+LOGGER = logging.getLogger(__name__)
+
+
+class KafkaPubSubClient(BasePubSubClient):
+    """Kafka client"""
+
+    def __init__(self, publisher_def):
+        """
+        Initialize object
+
+        :param publisher_def: provider definition
+
+        :returns: pygeoapi.pubsub.kafka.KafkaPubSubClient
+        """
+
+        super().__init__(publisher_def)
+        self.name = 'Kafka'
+        self.type = 'kafka'
+        self.sasl_mechanism = publisher_def.get('sasl.mechanism', 'PLAIN')
+        self.security_protocol = publisher_def.get('security.protocol', 'SASL_SSL')  # noqa
+
+        msg = f'Initializing to broker {self.broker_safe_url} with id {self.client_id}'  # noqa
+        LOGGER.debug(msg)
+
+    def connect(self) -> None:
+        """
+        Connect to an Kafka broker
+
+        :returns: None
+        """
+
+        args = {
+            'bootstrap_servers': f'{self.broker_url.hostname}:{self.broker_url.port}',  # noqa
+            'client_id': self.client_id,
+            'value_serializer': lambda v: to_json(v).encode('utf-8')
+        }
+        if None not in [self.broker_url.username, self.broker_url.password]:
+            args.update({
+                'security.protocol': self.security_protocol,
+                'sasl.mechanism': self.sasl_mechanism,
+                'sasl.username': self.broker_url.username,
+                'sasl.password': self.broker_url.password
+            })
+
+        LOGGER.debug('Creating Kafka producer')
+        try:
+            self.producer = KafkaProducer(**args)
+        except errors.NoBrokersAvailable as err:
+            raise PubSubClientConnectionError(err)
+
+    def pub(self, channel: str, message: str) -> bool:
+        """
+        Publish a message to a broker/channel
+
+        :param channel: `str` of topic
+        :param message: `str` of message
+
+        :returns: `bool` of publish result
+        """
+
+        LOGGER.debug(f'Publishing to broker {self.broker_safe_url}')
+        LOGGER.debug(f'Channel: {channel}')
+        LOGGER.debug(f'Message: {message}')
+        LOGGER.debug('Sanitizing channel for HTTP')
+        channel = channel.replace('/', '-')
+        channel = channel.replace(':', '-')
+        LOGGER.debug(f'Sanitized channel for Kafka: {channel}')
+
+        self.producer.send(channel, value=message)
+        self.producer.flush()
+
+    def __repr__(self):
+        return f'<HTTPPubSubClient> {self.broker_safe_url}'

pygeoapi/pubsub/mqtt.py

@@ -39,7 +39,7 @@
 class MQTTPubSubClient(BasePubSubClient):
     """MQTT client"""
 
-    def __init__(self, broker_url):
+    def __init__(self, publisher_def):
         """
         Initialize object
 
@@ -48,7 +48,7 @@ def __init__(self, broker_url):
         :returns: pycsw.pubsub.mqtt.MQTTPubSubClient
         """
 
-        super().__init__(broker_url)
+        super().__init__(publisher_def)
         self.type = 'mqtt'
         self.port = self.broker_url.port
 

requirements-pubsub.txt

@@ -1 +1,2 @@
+kafka-python
 paho-mqtt