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
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
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:
exaloop --with kafka build -release -o program program.codon
./program
Use an individual SDK without the CLI
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 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:
exaloop --with kafka build -release -o produce produce.codon
./produce localhost:9092 samples 100000
Consume and Commit
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
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:
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 and transformation into the same directory:
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 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 as
above. Provision that schema under trades-value in your registry, then create
codecs for your application:
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 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.
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:
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 covers
destination configuration and staging checks.
Next Steps
- Encode and decode Avro payloads with generated record types.
- Generate Protobuf messages for structured event payloads.
- Profile the worker with representative batches and broker traffic.