-   [](https://platform.exaloop.io/)
-   Streaming
-   Kafka

# Kafka

Build native producers, consumers, and streaming transformations with `exaloop.streaming.kafka`. The SDK uses librdkafka for cluster communication, while Codon executes the application's parsing, validation, and computation.

This guide covers delivery reports, processed offsets, Avro payloads, schema registry integration, and Kafka transactions. Choose the delivery contract before adding retries or external effects to a processing loop.

**Before you begin:** install the [all-SDK distribution](https://platform.exaloop.io/start/distribution/) and create an `events` topic on a local development broker. Use your usual Kafka tools to create topics and manage permissions; this is a client integration, not a broker extension.

## Produce Records

```python title="program.codon"
import exaloop.streaming.kafka as kafka

with kafka.Producer("localhost:9092", {"compression.type": "lz4"}) as producer:
    identifier = producer.produce("events", b"42", key=b"sensor-7")
    for report in producer.flush(10.0):
        report.raise_for_error()
        print(report.id, report.partition, report.offset)
```

Save this as `program.codon`, then build it with your SDK:

```sh title="Terminal"
exaloop --with kafka build -release -o program program.codon
./program
```

Use an individual SDK without the CLI

```sh title="Terminal"
codon build -release -plugin /path/to/sdk/kafka -o program program.codon
```

`produce()` queues the record; the delivery report tells you whether it reached Kafka. The example waits for outstanding work with `flush()` and checks every report before printing its partition and offset.

For a long-running producer, call `poll()` regularly to collect delivery reports. If a flush times out, keep polling or retry the flush rather than sending the records again: they may already have reached the broker. The default producer enables idempotence; turning it off weakens delivery guarantees.

The downloadable [producer example](https://platform.exaloop.io/assets/files/produce-71ca0dde0c74729765ece3a4fdaa4b0c.codon/) handles backpressure and checks every delivery report. Save it as `produce.codon` to write numbered UTF-8 values to partition zero of an existing topic:

```sh title="Terminal"
exaloop --with kafka build -release -o produce produce.codon
./produce localhost:9092 samples 100000
```

## Consume and Commit

```python title="Source"
import exaloop.streaming.kafka as kafka

with kafka.Consumer("localhost:9092", "event-workers") as consumer:
    consumer.subscribe(["events"])
    while True:
        batch = consumer.poll_batch(1024, 1.0)
        for message in batch:
            print(message.topic, message.partition, message.offset, message.value)
        if batch:
            consumer.commit(batch)
```

The consumer joins `event-workers`, reads available records, and commits only after processing the batch. A new group starts at the earliest available offset; an existing group resumes from its committed position. Use a different group ID when you want an independent reader.

A batch commit records the next offset after the highest included record in each partition. With concurrent processing, committing past an unfinished record can cause that record to be skipped when the consumer restarts. A failure between processing and committing can instead lead to redelivery, so external operations, such as database writes, need idempotency or deduplication.

`poll_batch()` may return fewer than the requested number of records, including an empty list. Polling also services consumer-group events, so it remains part of the worker loop when no application records are available.

### Partition Control

| Method | Behavior |
| --- | --- |
| `assign(partitions)` | Manual assignment, outside group partition balancing; `assign([])` clears it |
| `assignment()` | Current assigned topic partitions |
| `committed(partitions, timeout=10.0)` | Broker-stored offsets, or `OFFSET_INVALID` |
| `position(partitions)` | Next position after records delivered to the application, not a safe processed checkpoint |
| `seek(offsets, timeout=10.0)` | Reposition assigned partitions |
| `pause(partitions)` / `resume(partitions)` | Control fetching without relinquishing assignment |
| `watermarks(topic, partition, timeout=10.0)` | Query low and high offsets; high is the log end, not a record count |
| `offsets_for_times(partitions, timeout=10.0)` | Interpret each input offset as epoch milliseconds; return first matching offsets or `OFFSET_END` |

`TopicPartition(topic, partition, offset=OFFSET_STORED)` supports `OFFSET_BEGINNING`, `OFFSET_END`, and explicit nonnegative starting offsets for assignment and seeking. Commit and timestamp lookup require explicit nonnegative offsets. Do not mix subscribe and manual assignment on the same active consumer.

The downloadable [partition analytics example](https://platform.exaloop.io/assets/files/aggregate-625d2f2ab1cbbdbe96acbde73fd14ffd.codon/) replays a bounded range of numeric UTF-8 records and emits JSON count, sum, mean, and population variance without changing committed offsets. Save it as `aggregate.codon`:

```sh title="Terminal"
exaloop --with kafka build -release -o aggregate aggregate.codon
./aggregate localhost:9092 samples 0 0 100000
```

Adapt the per-batch calculation to decode events, run numerical code, or compute your own aggregates. When tuning it, measure the complete workload, including broker I/O, with representative records and batch sizes.

## Transform Avro Records

Use generated records for Kafka values, not Avro container files. Download the [trade schema](https://platform.exaloop.io/assets/files/trade-f77c6c8758199c4bc9ebdd8d6244ba86.avsc/) and [transformation](https://platform.exaloop.io/assets/files/avro_transform-639e741cd561cacea8abbcb0791c4b10.codon/) into the same directory:

```sh title="Terminal"
exaloop avrogen trade.avsc trade.codon
exaloop --with data --with kafka build -release -o avro_transform avro_transform.codon
./avro_transform localhost:9092 trades normalized-trades trade-normalizers 1000
```

Create both topics first. Populate `trades` with raw `Trade.to_bytes()` values using that schema. The worker uppercases each trade's symbol, preserves the other fields, keys, and headers, and forwards tombstones unchanged. Kafka routes output by key; input partition numbers and timestamps are not copied. It stops after 1,000 input messages, including tombstones, or fails after 60 seconds without input. Use a dedicated consumer group when running this bounded example.

Each batch is committed only after all output delivery reports succeed. This is **at-least-once**, not exactly-once: a crash or commit failure after successful delivery can cause duplicate output when the worker restarts. A decoding or delivery failure stops the worker without committing that batch; output already delivered from the batch may still be present. Correct or quarantine malformed input before restarting. The [transaction API](https://platform.exaloop.io/streaming/kafka/#transactions) provides atomic Kafka output and offset commits when that stronger contract is required.

All participants must agree on the writer schema. These payloads contain neither a schema ID nor registry framing. For registry-backed topics, use the codecs below instead of calling `Trade.from_bytes()` and `trade.to_bytes()` directly.

## Use a Schema Registry

`exaloop.streaming.schema_registry` supports the Confluent-compatible Avro REST API and wire format: a zero magic byte, a four-byte big-endian schema ID, then one raw Avro record. Build with both `--with data` and `--with kafka`. Registry HTTP requests and Avro encoding run natively; Python is not required by the application.

Generate `trade.codon` from the [trade schema](https://platform.exaloop.io/assets/files/trade-f77c6c8758199c4bc9ebdd8d6244ba86.avsc/) as above. Provision that schema under `trades-value` in your registry, then create codecs for your application:

```python title="Source"
import os
from trade import Trade
from exaloop.streaming.schema_registry import RegistryClient, AvroSerializer, AvroDeserializer

registry = RegistryClient(
    os.environ["SCHEMA_REGISTRY_URL"],
    username=os.environ.get("SCHEMA_REGISTRY_USERNAME", ""),
    password=os.environ.get("SCHEMA_REGISTRY_PASSWORD", ""),
    ca_file=os.environ.get("SCHEMA_REGISTRY_CA_FILE", ""),
)
encoder = AvroSerializer[Trade](registry, "trades-value")
decoder = AvroDeserializer[Trade](registry)
payload = encoder.encode(Trade(42, "ACME", i32(10), 12.5))
assert decoder.decode(payload).id == 42
```

Pass `payload` to `producer.produce()` as the value. On consumption, pass non-null `message.value` bytes to `decoder.decode()`. A Kafka tombstone stays `None`; do not frame or decode it. Keys can use a separate serializer and subject, commonly `<topic>-key`; values commonly use `<topic>-value`. Subject names are explicit, not inferred from the topic or record name.

The serializer looks up the generated record's exact schema during construction and retains its ID. It does not select the latest schema or register automatically. For an application authorized to create schema versions, either call `registry.register("trades-value", Trade._avro_schema())` explicitly or pass `auto_register=True` to the serializer. The registry's compatibility policy still applies; a rejected registration raises `RegistryError`.

The deserializer fetches each message's writer schema by ID and prepares a decoder for the generated reader type. This uses the same supported [schema evolution rules](https://platform.exaloop.io/data/avro/#generated-records) as raw Avro decoding. Malformed framing, unsupported schemas, incompatible evolution, and trailing datum bytes are rejected. Keep the existing delivery checks and commit only after processing succeeds; registry failures must not turn into committed, silently skipped records.

### Registry API

| API | Behavior |
| --- | --- |
| `RegistryClient(url, ...)` | Registry base URL, optionally including a path prefix; no embedded credentials, query, or fragment |
| `lookup(subject, schema: Schema) -> int` | Find the registered ID of an existing Avro schema; does not create a version |
| `register(subject, schema: Schema) -> int` | Explicitly register an Avro schema and return its ID |
| `get_schema(schema_id) -> Schema` | Fetch and cache a writer schema by unsigned 32-bit ID |
| `get_version(subject, version=-1) -> SchemaVersion` | Fetch a numbered version or latest (`-1`); returns `subject`, `version`, `schema_id`, and `schema` |
| `clear_cache()` | Clear the client's schema and lookup caches |
| `AvroSerializer[Record](registry, subject, auto_register=False, max_bytes=67108864)` | Prepare a generated-record encoder; `encode(record)` returns owned framed bytes |
| `AvroDeserializer[Record](registry, max_bytes=67108864)` | `decode(message)` returns a generated record; `clear_cache()` clears its prepared decoders |
| `frame(schema_id, payload, max_bytes=67108864)` / `unframe(message, max_bytes=67108864)` | Low-level framing only; `unframe` returns `(schema_id, payload)` without fetching or validating the schema |

Connection options are separate from Kafka's SASL/TLS settings:

| Option | Default and meaning |
| --- | --- |
| `username`, `password` | Empty; HTTP Basic authentication when a username is supplied |
| `bearer_token` | Empty; use instead of Basic authentication |
| `ca_file` | Empty; use libcurl's default trust store, or provide a CA certificate file |
| `timeout` | 10 seconds per request, including connection; positive, at most one hour |
| `max_response_bytes` | 8 MiB of decoded HTTP response body; configurable from 1 byte to 16 MiB |
| `cache_capacity` | 64 entries per cache, configurable from 1 to 1,024 |

Use HTTPS for production. Certificate and hostname verification remain enabled; redirects are not followed. Requests are synchronous and are not automatically retried. `RegistryError.status_code` is the HTTP status, or zero for transport, TLS, timeout, and response-limit failures. `error_code` contains the registry's integer error code when available, otherwise zero. Invalid successful responses and Avro validation failures raise `ValueError`.

Successful schema-ID and subject/schema lookups are cached. Prepared decoders have a separate cache with the same entry limit. Full caches evict the oldest inserted entry; version queries, registrations, and failures are not cached. `max_bytes` includes the five-byte envelope and accepts 5 bytes through 1 GiB. Schema text is limited to 1 MiB of UTF-8. These are entry and encoded-byte limits, not a total memory budget. Use clients and codecs from one application thread at a time, and do not mutate their configuration or cached `Schema` objects.

External schema references, Protobuf/JSON Schema framing, GUID/header-based schema identification, compatibility administration, OAuth token refresh, and mutual-TLS client certificates are not supported by this API.

## Transactions

A transactional producer has a stable `transactional.id` that identifies one logical worker. Initialization with `init_transactions(timeout=30.0)` establishes that identity before the first `begin_transaction()`. A replacement instance using the same ID fences its predecessor, which is why concurrently active workers need different IDs.

Within an active transaction, `send_offsets_to_transaction(consumer, processed_batch)` adds processed input offsets to the transaction. `commit_transaction()` then commits those offsets atomically with the produced output in the same Kafka cluster. A separate `consumer.commit()` would break that relationship. Read-committed consumers do not expose aborted output.

```python title="Source"
import exaloop.streaming.kafka as kafka

with kafka.Consumer("localhost:9092", "transform-workers") as consumer:
    consumer.subscribe(["input"])
    with kafka.Producer("localhost:9092", {"transactional.id": "transform-worker-0"}) as producer:
        producer.init_transactions()
        batch = consumer.poll_batch(1024, 10.0)
        while not batch:
            batch = consumer.poll_batch(1024, 10.0)
        producer.begin_transaction()
        for message in batch:
            producer.produce("output", message.value, key=message.key)
        producer.send_offsets_to_transaction(consumer, batch)
        for report in producer.commit_transaction():
            report.raise_for_error()
```

This example stops on a failure and attempts to abort an active transaction when leaving the producer block. A production worker needs a recovery policy based on `KafkaError`:

| Error flag | Recovery |
| --- | --- |
| `retriable` | Retry the failed transaction operation |
| `abort_required` | Abort, then replay the input from committed offsets |
| `fatal` | Replace the producer |

A timed-out commit may still be in progress, so retry that commit rather than starting a new transaction. Aborting does not rewind the consumer. Kafka transactions also do not cover HTTP calls, database writes, or other external effects.

## Connection Errors

Global broker, authentication, and TLS errors are retained even when logging is disabled. Consumer `poll()` and `poll_batch()` raise them as `KafkaError` rather than repeatedly reporting idle input. If a poll has already fetched valid messages, it returns that batch first and leaves the error for the next poll. No fetched records are discarded to report a global error.

Both clients also provide `check_errors()`: it raises the oldest retained global error, or returns normally when none is queued. It does not perform network I/O; poll regularly to service callbacks. Producers report global diagnostics through this method separately from `DeliveryReport` results, so an informational connection error cannot discard delivery reports or turn a successful transaction commit into an apparent failure. Call it after servicing producer polls and inspect delivery reports independently.

A global diagnostic is not proof that a particular message failed or that an operation is safe to retry. Unless `fatal` is set, librdkafka may recover the connection automatically. Transaction-specific `retriable` and `abort_required` flags belong to errors from transaction operations, not these global events. Each client retains at most 64 pending global errors; excess events are dropped until queued errors are consumed. Do not rely on this bounded queue as an audit log of every connection attempt.

## Values and Lifetimes

Send bytes or UTF-8 strings as keys and values. A `None` value is a Kafka tombstone, not an empty message; received keys and values are therefore `Optional[bytes]`. Values you have received remain usable after the next poll or client close.

Clients are used by one application thread at a time. A `with` block provides deterministic cleanup, including delivery flushing or consumer-group shutdown; garbage collection does not provide the same timing guarantees.

## Security and Deployment

Pass librdkafka string-valued properties in `config`. Set `bootstrap.servers` and `group.id` through the constructor arguments, not again in `config`. The default transport is **plaintext**, intended for a local broker. For production, configure verified TLS and your cluster's SASL mechanism, for example:

```python title="Source"
import os
import exaloop.streaming.kafka as kafka

config = {
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": os.environ["KAFKA_USERNAME"],
    "sasl.password": os.environ["KAFKA_PASSWORD"],
    "ssl.ca.location": os.environ["KAFKA_CA_FILE"],
}
producer = kafka.Producer(os.environ["KAFKA_BROKERS"], config)
producer.close()
```

Keep certificate and hostname verification enabled. The authentication mechanisms available depend on your librdkafka build. Creating a client is not itself a connection test; check polling and delivery errors to confirm it can reach the cluster.

The executable requires `libcodon_kafka`, librdkafka, libcurl and their dependencies, and matching Codon runtime libraries. Registry Avro codecs also require the Data SDK runtime. It does not require Java or the CLI's Python environment. For a configured project target, `exaloop --target app package --bundle-runtime` collects native dependencies; the [deployment guide](https://platform.exaloop.io/production/) covers destination configuration and staging checks.

## Next Steps

-   [Encode and decode Avro payloads](https://platform.exaloop.io/data/avro/#generated-records) with generated record types.
-   [Generate Protobuf messages](https://platform.exaloop.io/data/protobuf/) for structured event payloads.
-   [Profile the worker](https://platform.exaloop.io/start/profiling/) with representative batches and broker traffic.

Source: https://platform.exaloop.io/streaming/kafka/