# Migrating from confluent-kafka

> A drop-in Producer/Consumer shim that runs confluent-kafka code on pyrula.

Source: https://docs.pyrula.io/kafka/confluent-compat/

`pyrula.kafka.compat.confluent` mirrors the
[`confluent_kafka`](https://github.com/confluentinc/confluent-kafka-python) `Producer`,
`Consumer`, `AdminClient` and Schema Registry serializers, so existing code moves over with
an import change. What holds through the shim is on the
[client guarantees](https://docs.pyrula.io/kafka/guarantees.md) page.

```python
# before
from confluent_kafka import Producer, Consumer, TopicPartition, OFFSET_BEGINNING
```

```python
# after
from pyrula.kafka.compat.confluent import Producer, Consumer, TopicPartition, OFFSET_BEGINNING
```

New code should use the native [`KafkaProducer`](https://docs.pyrula.io/kafka/producer.md) and
[`KafkaConsumer`](https://docs.pyrula.io/kafka/consumer.md), which expose the batch, Arrow and exactly-once APIs.

## Config

The constructors take confluent's string-keyed dict and translate it to `KafkaConfig` /
`KafkaConsumerConfig`. An unknown key raises `KafkaException` (`_INVALID_ARG`). A key the
shim can't honour is refused the same way when ignoring it would change data, placement,
durability or security, and otherwise accepted with one `WARNING` per process.
[Confluent config keys](https://docs.pyrula.io/kafka/confluent-config-keys.md) lists every key and its verdict.

An omitted key gets confluent's default:

| Key | Shim default | Native default |
|---|---|---|
| `enable.auto.commit` | `True` | `False` |
| `auto.offset.reset` | `latest` | `earliest` |
| `session.timeout.ms` | `45000` | `10000` |
| `partitioner` | `consistent` (CRC32) | `murmur2` |
| `message.max.bytes` (producer) | `1000000` | 1 MiB |

To see the native config a dict becomes:

```python
from pyrula.kafka.compat.confluent import (
    consumer_config_from_confluent, producer_config_from_confluent,
)

producer_config_from_confluent({"bootstrap.servers": "localhost:9092", "acks": "all"})
consumer_config_from_confluent({"bootstrap.servers": "localhost:9092", "group.id": "g"})
```

## Producer

```python
p = Producer({"bootstrap.servers": "localhost:9092"})

def on_delivery(err, msg):
    if err is not None:
        log.error("delivery failed: %s", err)
    else:
        log.info("delivered to %s", msg.topic())

for value in values:
    p.produce("orders", value=value, key=b"k", callback=on_delivery)
p.flush()
```

`Producer()` connects in the background. Delivery callbacks and `error_cb` fire on `poll()`
and `flush()`, as in confluent, so a producer that never polls gets `BufferError` once the
queue fills. Call `flush()` before exit; unserved reports are dropped when the `Producer` is
garbage collected.

### Transactions

```python
p.init_transactions()
p.begin_transaction()
try:
    p.produce("orders", value=v, key=k, on_delivery=report)
    p.send_offsets_to_transaction(positions, consumer.consumer_group_metadata())
    p.commit_transaction()
except KafkaException as exc:
    if exc.args[0].txn_requires_abort():
        p.abort_transaction()   # producer stays usable
    else:
        raise
```

`commit_transaction()` flushes first and raises with `txn_requires_abort()` `True` if any
record failed. A `produce()` with no open transaction raises `KafkaException(_STATE)`. A
commit, abort or init whose `timeout` runs out raises a retriable error and keeps going;
call it again to resume.

## Consumer

```python
c = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "orders-worker",
    "auto.offset.reset": "earliest",
})
c.subscribe(["orders"])
while running:
    msg = c.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        log.error("poll error: %s", msg.error())
        continue
    handle(msg.value())
    c.commit(msg)
c.close()
```

`Consumer()` connects in the background. Until a broker has answered, `poll()` returns
`None` after its timeout, and connect failures go to `error_cb` (or the `ERROR` log). Poll
errors come back as a `Message` whose `error()` is a `KafkaError`, never as an exception.
`consume(num_messages, timeout)` returns a list of messages the same way.

### Offsets

```python
c.commit(offsets=[TopicPartition("orders", 0, 1234)])   # absolute next-read position
c.assign([TopicPartition("orders", 0, 1234), TopicPartition("orders", 1, OFFSET_BEGINNING)])
c.seek(TopicPartition("orders", 0, OFFSET_END))
tps = c.assignment()
positions = c.position(tps)
committed = c.committed(tps)
```

With auto-commit on, commits send the offset of the last message `poll()` returned. With
`enable.auto.offset.store=false`, only `store_offsets()` advances commits; stage before the
next `poll()`.

`TopicPartition` compares, hashes and sorts on `(topic, partition)` alone, as in confluent,
so `TopicPartition("t", 0, 5) == TopicPartition("t", 0, 7)`.

### Callbacks

`subscribe(topics, on_assign=, on_revoke=, on_lost=)` takes confluent's rebalance callbacks,
called as `callback(consumer, partitions)`. Offsets staged with `store_offsets` are committed
for revoked partitions after `on_revoke` returns. `commit()` inside `on_lost` raises
`KafkaException`.

`on_commit(err, partitions)` reports the default asynchronous `commit()` and each automatic
commit, from `poll()`, `consume()` and `close()`. A `commit(asynchronous=False)` returns or
raises instead.

### Partition EOF

With `enable.partition.eof=true`, `poll()` and `consume()` return a message with
`error().code() == KafkaError._PARTITION_EOF` at the end of a partition. `commit(message=)`
and `store_offsets(message=)` refuse it.

### Threads

Other threads can call `commit`, `store_offsets`, `seek`, `pause` and the rest while one
thread polls. A second concurrent `poll()` raises `KafkaException` with `_CONFLICT`.
`close()` from another thread ends the running poll with `None`.

## Admin

```python
from pyrula.kafka.compat.confluent.admin import AdminClient, NewTopic

admin = AdminClient({"bootstrap.servers": "localhost:9092"})
futures = admin.create_topics(
    [NewTopic("orders", num_partitions=6, replication_factor=3,
              config={"cleanup.policy": "compact"})],
    operation_timeout=30,
)
for topic, future in futures.items():
    try:
        future.result()
    except KafkaException as exc:
        if exc.args[0].code() != KafkaError.TOPIC_ALREADY_EXISTS:
            raise
```

`create_topics`, `delete_topics`, `create_partitions` and `list_topics` return a dict of
already-resolved futures per topic. `NewTopic(replica_assignment=...)` raises
`NotImplementedError`.

## Schema Registry serializers

`SerializingProducer`, `DeserializingConsumer` and the Avro and JSON Schema serializers use
confluent's import paths:

```python
from pyrula.kafka.compat.confluent import SerializingProducer, DeserializingConsumer
from pyrula.kafka.compat.confluent.schema_registry import SchemaRegistryClient
from pyrula.kafka.compat.confluent.schema_registry.avro import AvroSerializer, AvroDeserializer
from pyrula.kafka.compat.confluent.schema_registry.json_schema import JSONSerializer, JSONDeserializer
from pyrula.kafka.compat.confluent.serialization import StringSerializer, StringDeserializer

registry = SchemaRegistryClient({"url": "http://localhost:8081"})

p = SerializingProducer({
    "bootstrap.servers": "localhost:9092",
    "key.serializer": StringSerializer("utf_8"),
    "value.serializer": AvroSerializer(registry, ORDER_SCHEMA),
})
p.produce("orders", key="o-1", value={"id": 1, "sku": "sku-1", "qty": 2})
p.flush()

c = DeserializingConsumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "g",
    "key.deserializer": StringDeserializer("utf_8"),
    "value.deserializer": AvroDeserializer(registry, ORDER_SCHEMA),
})
c.subscribe(["orders"])
msg = c.poll(1.0)          # msg.key() -> "o-1", msg.value() -> the dict
```

They wrap the native [`AvroRegistrySerde` / `JsonSchemaRegistrySerde`](https://docs.pyrula.io/kafka/formats.md) and
speak the Confluent wire format. Schemas are registered under `<topic>-key` /
`<topic>-value` on first serialize.

Not supported:

- `subject.name.strategy` other than `topic_subject_name_strategy` (`NotImplementedError`).
- `auto.register.schemas=False` (`NotImplementedError`); other serializer `conf` keys such as
  `use.latest.version` raise `ValueError`.
- `AvroDeserializer(return_record_name=True)` (`NotImplementedError`).
- `DeserializingConsumer.consume()` (`NotImplementedError`); use `poll()`.
- `ProtobufSerializer` and `ProtobufDeserializer` (`NotImplementedError` on construction).
  The native `ProtobufRegistrySerde` works.

`JSONSerializer` does not validate documents against the schema.

## Differences from librdkafka

- The default asynchronous `commit()` does not raise its failure, as in confluent. A failure
  goes to `on_commit`, or is logged at `WARNING` when there is none.
- `msg.timestamp()` never reports `TIMESTAMP_LOG_APPEND_TIME`.
- Keyless records are sticky under every `partitioner` spelling. librdkafka sends them all to
  one partition under `consistent` and `murmur2`.
- `request.timeout.ms` bounds broker requests on a consumer (30 s default); librdkafka
  ignores it there.
- Confluent's `fnv1a`, `fnv1a_random` and `random` partitioners are refused.
- Partition EOF messages have `value()` of `None`, where confluent's is the error text.
- Statistics carry librdkafka's top-level counters only; per-broker, per-topic and group
  sections are absent.
- `consumer_group_metadata()` is `None` before the first connection.