Apache Kafka

The akafka subpackage (hexkit.providers.akafka) implements hexkit’s event protocols on top of Apache Kafka, using the asynchronous aiokafka client library. It is hexkit’s production event-streaming backend; the architectural background is covered in the Event-Driven Architecture chapter.

The subpackage provides:

Two hybrid providers combine Kafka with MongoDB and are documented in the MongoDB + Kafka chapter: the outbox publisher (MongoKafkaDaoPublisher) and the PersistentKafkaPublisher.

Install hexkit with the akafka extra to use these providers:

pip install hexkit[akafka]

Configuration

All Kafka providers are configured through a single Pydantic settings class, KafkaConfig (hexkit.providers.akafka.KafkaConfig). The connection-related parameters, relevant to every Kafka provider, are:

Parameter Default Description
service_name required Name of the service, e.g. "user-registry". Doubles as the Kafka consumer group ID, so all instances of a service share the consumption workload.
service_instance_id required Identifier unique to this instance of the service, e.g. "instance-001". The Kafka client ID is derived as <service_name>.<service_instance_id>.
kafka_servers required List of connection strings for the Kafka bootstrap servers, e.g. ["localhost:9092"].
kafka_security_protocol "PLAINTEXT" Either "PLAINTEXT" or "SSL".
kafka_ssl_cafile "" Path to the certificate authority file used to sign the broker certificates. If empty, the system CA is used if OpenSSL finds one.
kafka_ssl_certfile "" Optional client certificate file (including any CA certificates needed to establish its authenticity).
kafka_ssl_keyfile "" Optional client private key file.
kafka_ssl_password "" Optional password for the client private key.
kafka_max_message_size 1048576 (1 MiB) The largest transmittable message size in bytes, applied to producers (before compression) and consumers alike.

KafkaConfig carries further parameters that only concern one side of the event flow; they are documented where they apply: compression and correlation ID generation on the publisher page, and retries and the dead letter queue on the subscriber page.

Providing Configuration Values

Since KafkaConfig is a pydantic-settings BaseSettings class, every parameter can be supplied as an environment variable named after the field (case-insensitive; list values as JSON):

export SERVICE_NAME="user-registry"
export SERVICE_INSTANCE_ID="instance-001"
export KAFKA_SERVERS='["kafka-server-1:9092", "kafka-server-2:9092"]'

In a typical service, KafkaConfig is not instantiated on its own. Instead, the service defines one config class inheriting from all the hexkit config classes it needs (plus its own settings) and decorates it with hexkit’s config_from_yaml, which layers the configuration sources — keyword arguments override environment variables, which override secret files under /secrets, which override a config YAML, which overrides the defaults:

from hexkit.config import config_from_yaml
from hexkit.providers.akafka import KafkaConfig


@config_from_yaml(prefix="my_service")
class Config(KafkaConfig):
    """Configuration for my service."""

    ...  # further settings, e.g. from other hexkit config classes


config = Config()  # reads .my_service.yaml, MY_SERVICE_* env vars, etc.

With a prefix set, environment variables take the form MY_SERVICE_KAFKA_SERVERS='["localhost:9092"]', and the config YAML is looked up via the MY_SERVICE_CONFIG_YAML environment variable or as .my_service.yaml in the current or home directory.

Kafka Test Utils

The hexkit.providers.akafka.testutils module provides pytest fixtures that spin up a disposable Kafka broker in a container, so integration tests exercise real broker semantics without external infrastructure — publish test events, record what a service publishes, and assert expectations. The Kafka Test Utils page documents the fixtures, the KafkaFixture utility methods, and common testing patterns, including in-memory alternatives for broker-less unit tests.