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.

Messaging observability

kafka_pubsub owns Kafka transport lifecycle observability, not application or business lifecycle logging. It never logs message values, keys, credentials, or full client configuration. At INFO level it logs safe lifecycle events using topic, partition, and offset where Kafka makes them available:

• publication initiation, followed by either broker-confirmed publication or enqueue, delivery, or timeout failure;
• synchronous offset-commit initiation, followed by broker-confirmed commit or commit failure; and
• RETRY as an explicitly unacknowledged message, with no commit attempted.

Successful producer calls return DeliveryMetadata(topic, partition, offset) only after the delivery callback confirms publication. Successful ACK and TERMINAL calls return CommitResult(topic, partition, offset, disposition) only after the synchronous commit returns. Applications should retain or log these returned values beside their own request/job identifiers to correlate their lifecycle with Kafka; package logs cannot infer application identity. Message receipt context is available at DEBUG level. The package does not log per-poll idle activity.

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.