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

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 and decorates it with config_from_yaml to draw the values from environment variables, secret files, and a config YAML:

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.

See the Configuration chapter for the full precedence of configuration sources and the role of the prefix.

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.