Consumer

KafkaConsumer takes a KafkaConsumerConfig and polls batches. Use it as a context manager so it leaves the group cleanly.

from pyrula.kafka import KafkaConsumerConfig, KafkaConsumer

config = KafkaConsumerConfig(
    bootstrap_servers="localhost:9092",
    topics=["events"],
    group_id="my-group",
)

with KafkaConsumer(config) as consumer:
    while True:
        result = consumer.poll_batch(max_records=500, timeout_ms=1000)
        if result.is_err():
            handle(result.error)
            continue
        batch = result.value
        if batch.count == 0:
            continue
        for record in batch.records.to_list():
            process(record)
        consumer.commit(batch).unwrap_or_raise()

poll_batch returns an Either. A batch carries its offsets, so commit(batch) needs nothing else.

For one record at a time, consume(consumer) yields KafkaRecords from poll_batch forever. A poll error raises ConsumerError, and you commit by hand.

A tombstone reads back with record.value None. poll_bytes_batch / poll_bytes_packed cannot express null: a drain that meets a tombstone returns Err(PollError) naming the record and advances no offset. Read compacted topics through poll_batch.

Committing offsets

Auto-commit is off by default (enable_auto_commit=False). Commit after processing a batch for at-least-once:

consumer.commit(batch).unwrap_or_raise()  # synchronous
consumer.store_offsets(batch)             # stage; the next commit sends these instead of the fetch position

commit_offsets(offsets) commits an explicit list. A negative offset in it (or in store_offsets) is refused with CommitError, except OFFSET_INVALID (-1001, a partition that has not started), which is skipped.

close(), unsubscribe(), unassign() and a replacing subscribe() commit offsets staged with store_offsets, after on_revoke has run. The fetch position commits too only when enable_auto_commit is True.

Assignment and position

consumer.subscribe(["events"], on_assign=..., on_revoke=..., on_lost=...)
consumer.assign([TopicPartition("events", 0)])

A manual assignment doesn’t join the group, but with group_id set it commits to it. Each partition starts at the committed offset, falling back to auto_offset_reset. Pass assign(partitions, start_offsets={tp: offset}) (absolute offset, OFFSET_BEGINNING or OFFSET_END) to skip the committed offset. Without a group_id, commit() returns a CommitError.

A subscribed topic that does not exist yet is waited for. One the client may not read (TopicAuthorizationFailed) returns a PollError naming the topic, once, and polling continues on the others.

A topic list takes exact names. To match by pattern pass the pattern alone, as a str or a re.Pattern (consumer.subscribe("^events-.*")); a list entry starting with ^ is a ConfigError, in subscribe and in KafkaConsumerConfig(topics=...) alike. The confluent shim accepts ^ entries in a list as librdkafka does.

Move with seek, seek_to_beginning, seek_to_end; inspect with position, committed, get_watermark_offsets, lag; apply backpressure with pause and resume; offsets_for_times looks up by timestamp. get_watermark_offsets and offsets_for_times work for unassigned partitions and take an optional timeout_ms. position is the offset after the last record delivered to you.

from pyrula.kafka import OFFSET_BEGINNING, OFFSET_END, OFFSET_STORED, OFFSET_INVALID

consumer.seek(TopicPartition("events", 0), OFFSET_BEGINNING)

seek takes an offset >= 0 or one of those constants; any other negative value is a ConsumerError.

Rebalance callbacks

A callback is called as cb(consumer, partitions) and runs with the consumer released, so the confluent idioms work: commit() in on_revoke, seek in on_assign, and position, committed, pause, resume anywhere. The rebalance is acknowledged when the callback returns; explicit acknowledgement also works (assign / unassign under eager, incremental_assign / incremental_unassign under cooperative-sticky), naming exactly the partitions you were given.

subscribe, unsubscribe, close and every poll_* are refused inside a callback on its thread with RebalanceError, as is commit inside on_lost. Offsets staged with store_offsets are committed for revoked partitions after on_revoke. Callback time counts toward max_poll_interval_ms. Under AsyncKafkaConsumer the callback receives the sync KafkaConsumer; don’t await the wrapper from inside it.

Threads

A KafkaConsumer can be shared across threads with one poller at a time.

• While one thread polls, every other call (commit, store_offsets, position, seek, pause, resume, assignment, …) is served, so committing from a worker thread while the main thread polls works.
• A second concurrent poll_*() raises ConsumerError (_conflicting_use).
• close() from another thread wakes the poll, commits staged offsets and leaves the group. Later calls except metrics() raise RuntimeError.
• Dropping without close() does not leave the group; partitions move after session_timeout_ms.

Callback	Runs on	May call the consumer?
on_assign / on_revoke / on_lost	the thread whose call drove the rebalance	yes, minus the refusals above
stats_cb, and error_cb / logger raised by a call	that call’s thread	yes, except a nested poll_*() or close()
error_cb / logger for background errors	a client thread	no: every call raises _conflicting_use; hand the work to another thread

subscribe, unsubscribe, unassign and close from another thread wait for a running rebalance callback, so a callback that blocks on such a thread deadlocks, as in librdkafka.

Static membership

Set group_instance_id to a stable, unique id per instance (KIP-345) so a rolling restart does not trigger rebalances:

config = KafkaConsumerConfig(
    bootstrap_servers="localhost:9092",
    topics=["events"],
    group_id="my-group",
    group_instance_id="worker-1",   # stable per instance (e.g. pod ordinal)
)

The same id rejoining before session_timeout_ms keeps its partitions. Requires Kafka 2.3+.

Async

AsyncKafkaConsumer mirrors the sync API. consume(max_records=, timeout_ms=) is an async generator of KafkaRecords.

from pyrula.kafka import AsyncKafkaConsumer

async with AsyncKafkaConsumer(config) as consumer:
    async for record in consumer.consume(max_records=500, timeout_ms=1000):
        process(record)

A poll cancelled after the fetch took records (asyncio.wait_for timing out, or task.cancel()) loses nothing: the next poll delivers the batch in order. The put-back is a rewind that refetches (about half a second on a local broker), so a wait_for shorter than a fetch can cancel every poll and make no progress. Use timeout_ms + 1 s.