Migrating from confluent-kafka

pyrula.kafka.compat.confluent mirrors the confluent_kafka 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 page.

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

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

New code should use the native KafkaProducer and KafkaConsumer, 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 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:

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

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

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

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

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

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:

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