benzene.kafka

Host Benzene handlers on Apache Kafka — a self-hosted consumer and a Kafka-produce outbound client. Distribution: benzene-kafka (depends only on benzene-core).

pip install benzene-kafka             # the binding + testing helpers
pip install "benzene-kafka[kafka]"    # + confluent-kafka for the real clients

Overview

Kafka is a queue/stream transport, so the binding follows the same shape as the SQS/SNS/Pub/Sub bindings (transport-bindings §"Kafka"):

• Topic — resolved from the record's topic header (the reserved metadata key; the cross-port convention of wire-contracts §2), falling back to the Kafka topic the record was consumed from when the header is absent, so a producer that writes no header still routes (topicSources in transport-metadata-cases.json); the header wins whenever present.
• Headers — the record's other Kafka headers, UTF-8 decoded, become the Benzene headers.
• Body — the record value, decoded as the UTF-8 JSON body.
• Scope — one record is one pipeline invocation and one DI scope (never per batch).
• Settlement — there is no response channel: the result settles the record (transport-bindings §1, obligation 8). By default the loop is skip-and-continue, the default a partition-ordered stream may choose: a record's offset is committed only when its invocation ended isSuccessful, and an unacknowledged record (a failure, a missing result, an unmatched topic) is passed over — not committed itself, but the next commit on its partition passes it, so it is redelivered only if the consumer restarts or rebalances first. The unit of settlement is the record offset, committed cumulatively per partition. run_consumer_loop(..., stop_on_failure=True) is the at-least-once mode: the loop raises MessageHandlingError at the first unacknowledged record without committing it, so the worker stops and a restart resumes there. KafkaConsumerApp(..., settlement=Settlement(at_most_once=True)) commits everything; Settlement(ignore_unmatched_topics=True) commits a topic no handler serves as ignored.
• Failure — a malformed body or a handler fault becomes a failure result, never an exception out of the loop, so a poison record cannot crash the host.

The binding is duck-typed against confluent-kafka, so decode, dispatch, and send run in memory with fakes — no broker, no SDK. Only the default consumer/producer clients need the [kafka] extra.

KafkaConsumerApp — inbound

from benzene.core import application_from, build_application
from benzene.kafka import KafkaConsumerApp, run_consumer_loop

definition, _ = build_application(OrdersStartUp)
app = KafkaConsumerApp(application_from(definition))

await app.handle_message(record)          # one record -> one invocation; returns the mapped Result
await run_consumer_loop(app, consumer)    # the self-hosted worker: poll -> dispatch -> commit

• KafkaConsumerApp(application, *, settlement=Settlement()) wraps a BenzeneMessageApplication; handle_message(record) decodes the record, runs the pipeline, and returns the Result (decoded with the envelope's isSuccessful authoritative) for the loop to settle.
• run_consumer_loop(app, consumer, *, poll_timeout=1.0, should_continue=..., commit=True, stop_on_failure=False, on_result=None) drives a duck-typed consumer (poll(timeout) → a record or None; commit(message=...)). A record carrying a broker error (record.error()) is skipped; with commit=True an acknowledged record's offset is committed; stop_on_failure turns the default skip-and-continue into at-least-once (above). should_continue bounds the loop (a real worker loops forever).
• build_kafka_consumer(*, bootstrap_servers, group_id, topics, auto_offset_reset="earliest", **config) constructs and subscribes a real confluent_kafka.Consumer configured to match the loop above — most importantly enable.auto.commit=False, which is what lets the loop decide what is committed — with auto-commit on, a record is committed before its outcome is known. Any **config (dotted confluent-kafka keys) wins over the defaults. Needs the [kafka] extra.
• kafka_consumer_worker(app, consumer, *, close=True, **loop_options) returns a benzene.core.WorkerHost leg wrapping the loop above, for a process that also serves HTTP. It is a closure over run_consumer_loop(..., should_continue=stop.should_continue) plus the finally: consumer.close() — reach for it only when there is more than one transport.
• decode_kafka_message(record) is the pure decode step (record → {topic, headers, body}), exposed for custom loops.

KafkaMessageSender — outbound

from benzene.kafka import KafkaMessageSender

sender = KafkaMessageSender("orders-events", bootstrap_servers="localhost:9092")
await sender.send_message("orders:created", order, headers={"x-correlation-id": "abc"})

Implements the benzene.core.MessageSender port over a Kafka producer: it serializes the message to the JSON body, forwards the header dictionary onto the record's Kafka headers (so correlation/trace propagation rides across the hop), and carries the Benzene topic in the topic header. All Benzene topics are produced to the one configured Kafka topic, header-routed. A produce/flush failure (or a delivery error reported to the callback) maps to service-unavailable — timeout for a raised timeout. Inject a producer for tests; otherwise a confluent_kafka.Producer is created lazily from bootstrap_servers.

Testing

benzene.kafka.testing provides a native-record builder and an in-memory test host, and the shared harness specializes to Kafka in one call:

from benzene.testing import create_test_host

host = create_test_host(OrdersStartUp).with_services(overrides).build_kafka()
result = await host.send_kafka("orders:place", body={"sku": "ABC", "quantity": 2})

RecordingKafkaConsumer (a replay consumer that records committed offsets) lets a test assert the loop's at-least-once behaviour without a broker. See the runnable examples/kafka_orders/.

Exports

KafkaConsumerApp, KafkaMessageSender, KafkaMessage, TOPIC_HEADER, build_kafka_consumer, decode_kafka_message, kafka_consumer_worker, run_consumer_loop; and from benzene.kafka.testing: KafkaTestHost, KafkaMessageBuilder, FakeKafkaMessage, RecordingKafkaConsumer.

See also

• Transport bindings — the language-neutral binding contract this implements.
• benzene.core — the registry, pipeline, and MessageSender port this builds on.

Edit this page on GitHub