Metadata-Version: 2.4
Name: kafka_pubsub
Version: 1.0.0
Summary: Apache Kafka customized Producer, Consumer, helper classes and methods
Author-email: Emmanuel Chukwu <emmanuel.chukwu@gmail.com>
License: MIT
Requires-Python: >=3.8
Description-Content-Type: text/markdown
Requires-Dist: confluent-kafka>=2.0
Requires-Dist: python-dotenv>=0.21
Provides-Extra: test
Requires-Dist: pytest>=7; extra == "test"

# kafka_pubsub

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

## Install

```bash
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](RELEASE.md).

## Consuming safely

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

```python
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:

```python
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.
