# Consumer

> Poll batches, commit offsets, and manage assignment and position.

Source: https://docs.pyrula.io/kafka/consumer/

`KafkaConsumer` takes a [`KafkaConsumerConfig`](https://docs.pyrula.io/kafka/configuration.md) and polls batches.
Use it as a context manager so it leaves the group cleanly.

```python
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`](https://docs.pyrula.io/primitives/either.md). A batch carries its offsets, so
`commit(batch)` needs nothing else.

For one record at a time, `consume(consumer)` yields `KafkaRecord`s 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:

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

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

```python
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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-345)) so a rolling restart
does not trigger rebalances:

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

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