kafka_pubsub

Small, generic wrappers around confluent-kafka for applications that need explicit Kafka delivery and acknowledgement boundaries.

Install

pip install kafka_pubsub

Runtime dependencies (confluent-kafka and python-dotenv) are declared by the package. For local tests use pip install -e '.[test]'.

API and releases

The current hardened API is intended to become the governed stable 1.0.0 boundary; it is materially breaking relative to the historical public 0.2.x series. The checked-in source intentionally reports the development version 1.0.0.dev0, not a public 0.2.x release. Release version calculation, tagged build overlays, Git releases, and separately manual PyPI publication are described in RELEASE.md.

Consuming safely

Subclass Consumer and make process(message) return a Disposition:

from kafka_pubsub.consumer import Consumer, Disposition

class Orders(Consumer):
    def process(self, message):
        # Complete every application side effect first.
        handle(message.value())
        return Disposition.ACK

ACK synchronously commits after process returns; its CommitResult is passed to on_message_success. TERMINAL also synchronously commits, but calls on_message_terminal; use it only after the application has deliberately handled a poison/terminal message (for example, by its own DLQ policy). RETRY does not commit. Exceptions and invalid dispositions do not commit and are sent to on_message_failure. Consumer enable.auto.commit and enable.auto.offset.store are always disabled. A failed synchronous commit is also sent to that hook and then raised from the runner.

Run consumers with a cooperative stop controller:

from kafka_pubsub.multi_consumer_runner import MultiConsumerRunner, StopController

controller = StopController()
# another thread/signal adapter may call controller.request_stop()
MultiConsumerRunner.run([Orders], controller=controller, drain_timeout=30)

A stop request prevents the next message from entering application work. The current synchronous handler is allowed to return, then initialized consumers are closed in finally. drain_timeout is a bounded runner lifetime mechanism; the package does not kill application work or impose a signal policy.

Producing

Producer.produce_and_wait() (and the compatibility spelling produce()) uses a bounded wait, raises ProducerError or ProducerTimeoutError on enqueue, delivery, or timeout failure, and returns DeliveryMetadata(topic, partition, offset) only after broker-confirmed delivery. kafka_produce() has the same semantics. Producers default to acks=all, retries, and idempotence where the installed librdkafka supports it; this is not exactly-once processing.

Topics and configuration

resolve_topic(topic) applies KAFKA_TOPICS_PREFIX to producer and consumer topics identically, exactly once. A topic already starting with the prefix is treated as resolved. kafka_topics_prefix() remains available.

Package settings use snake_case names from KAFKA_CONFIG; pass advanced confluent-kafka/librdkafka options either as dotted keys or in a client_options dictionary. Unknown non-dotted options fail early. A non-empty consumer_group is required, and automatic commit/offset-store settings are rejected to preserve the explicit acknowledgement contract.

Applications remain responsible for domain retry policy, idempotency, side effects, dead-letter handling, persistence, readiness, and process lifecycle.