# Quickstart

> Produce a batch, poll it back, and read the records.

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

`pip install "pyrula[kafka]"`. Calls work on batches and return
[`Either`](https://docs.pyrula.io/primitives/either.md) values instead of raising. The examples assume an `events`
topic: a local broker auto-creates it, managed Kafka usually needs it created first
([Admin](https://docs.pyrula.io/kafka/admin.md)).

## Produce

```python
from pyrula.kafka import KafkaConfig, KafkaProducer

config = KafkaConfig(bootstrap_servers="localhost:9092")

with KafkaProducer(config) as producer:
    receipt = producer.produce_many_bytes("events", values=[b"a", b"b", b"c"])
    report = receipt.get().unwrap_or_raise()   # blocks until the broker acks the batch
```

`produce_many_bytes` returns a `ProduceReceipt` at once; `.get()` blocks until the broker acks.

## Consume

```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():
            raise RuntimeError(result.error)
        batch = result.value
        if len(batch) == 0:
            continue          # early polls can be empty while the group joins
        for record in batch:
            print(record.partition, record.offset, record.value)
        consumer.commit(batch).unwrap_or_raise()
        break
```

A batch is iterable and sized, and carries the offsets to commit. A record has `topic`,
`partition`, `offset`, `key`, `value` (bytes), `headers` and `timestamp_ms`.

## Next

- [Producer](https://docs.pyrula.io/kafka/producer.md) and [Consumer](https://docs.pyrula.io/kafka/consumer.md)
- [Transactions](https://docs.pyrula.io/kafka/transactions.md) for exactly-once
- [Configuration](https://docs.pyrula.io/kafka/configuration.md)