# Apache Kafka

The `akafka` subpackage (`hexkit.providers.akafka`) implements hexkit's event protocols on top of [Apache Kafka](https://kafka.apache.org/), using the asynchronous [aiokafka](https://aiokafka.readthedocs.io/) client library. It is hexkit's production event-streaming backend; the architectural background is covered in the [Event-Driven Architecture](../../../user-guide/arch_concepts/event_driven_arch.md) chapter.

The subpackage provides:

- [`KafkaEventPublisher`](../../../user-guide/providers/kafka/publisher.md) -- publishes events, implementing the [`EventPublisherProtocol`](../../../reference/protocols.eventpub.EventPublisherProtocol.md).
- [`KafkaEventSubscriber`](../../../user-guide/providers/kafka/subscriber.md) -- consumes events and hands them to a service-defined translator implementing the [`EventSubscriberProtocol`](../../../reference/protocols.eventsub.EventSubscriberProtocol.md).
- [`KafkaOutboxSubscriber`](../../../user-guide/protocols/daosubscriber.md#running-a-subscriber-the-kafka-provider) -- a convenience wrapper for consuming outbox events with translators implementing the [`DaoSubscriberProtocol`](../../../user-guide/protocols/daosubscriber.md).
- [Kafka test utils](../../../user-guide/providers/kafka/test_utils.md) -- pytest fixtures and helpers for testing against a real, disposable Kafka broker.

Two hybrid providers combine Kafka with MongoDB and are documented in the [MongoDB + Kafka](../../../user-guide/providers/mongokafka.md) chapter: the outbox publisher (`MongoKafkaDaoPublisher`) and the [PersistentKafkaPublisher](../../../reference/providers.mongokafka.PersistentKafkaPublisher.md#hexkit.providers.mongokafka.PersistentKafkaPublisher).

Install hexkit with the `akafka` extra to use these providers:

``` bash
pip install hexkit[akafka]
```


# Configuration

All Kafka providers are configured through a single Pydantic settings class, [`KafkaConfig`](../../../reference/providers.akafka.KafkaConfig.md) ([hexkit.providers.akafka.KafkaConfig](../../../reference/providers.akafka.KafkaConfig.md#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](../../../reference/providers.akafka.KafkaConfig.md#hexkit.providers.akafka.KafkaConfig) carries further parameters that only concern one side of the event flow; they are documented where they apply: [compression and correlation ID generation](../../../user-guide/providers/kafka/publisher.md#configuration) on the publisher page, and [retries and the dead letter queue](../../../user-guide/providers/kafka/subscriber.md#configuration) on the subscriber page.


## Providing Configuration Values

In a typical service, [KafkaConfig](../../../reference/providers.akafka.KafkaConfig.md#hexkit.providers.akafka.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](../../../reference/config.config_from_yaml.md#hexkit.config.config_from_yaml) to draw the values from environment variables, secret files, and a config YAML:

``` python
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](../../../user-guide/configuration.md) 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](../../../user-guide/providers/kafka/test_utils.md) page documents the fixtures, the `KafkaFixture` utility methods, and common testing patterns, including in-memory alternatives for broker-less unit tests.