Consumer

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

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: Ok(batch) or Err(error). A batch knows its count and carries the offsets to commit, so you commit the batch directly rather than building offset lists by hand.

Committing offsets

Auto-commit is off by default (enable_auto_commit=False), so you control when offsets move. Commit after you’ve processed a batch for at-least-once:

consumer.commit(batch).unwrap_or_raise()  # synchronous commit of the batch's offsets
consumer.store_offsets(batch)     # stage offsets for the next auto/periodic commit

commit_offsets(offsets) commits an explicit offset list when you need finer control.

Assignment and position

Subscribe for group-managed assignment, or assign partitions yourself:

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

Move around with seek, seek_to_beginning, seek_to_end. Inspect with position, committed, get_watermark_offsets, and lag. Backpressure with pause and resume. offsets_for_times looks up offsets by timestamp.

The special offsets are exported as constants for use with seek and assign:

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

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

Static membership

By default a consumer is a dynamic group member: it leaves the group on shutdown, which triggers a rebalance, and a rolling restart of a fleet causes a rebalance storm. Set group_instance_id to a stable, unique id per instance to make it a static member (KIP-345):

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

A static member keeps its broker-side membership across a graceful restart: it does not send LeaveGroup, so the same group_instance_id rejoins as the same member before session_timeout_ms elapses, and its partitions are not reassigned. This avoids the rebalance churn that rolling restarts otherwise cause. Requires a broker that supports JoinGroup v5 (Kafka 2.3+); give each instance a distinct id.

Async

AsyncKafkaConsumer mirrors the sync API. consume(max_records=, timeout_ms=) is an async generator that yields KafkaRecord objects one at a time; poll_batch and commit are also available with the same semantics as the sync consumer.

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)