Migrating from confluent-kafka

pyrula.kafka.compat.confluent provides Producer and Consumer classes that mirror the confluent_kafka API, so existing code moves over with little more than an import change.

# before
from confluent_kafka import Producer, Consumer, TopicPartition

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

The constructors take the same string-keyed config dict ("bootstrap.servers", "group.id", "enable.idempotence", …); it is translated to the native KafkaConfig / KafkaConsumerConfig. Unrecognized keys raise a ValueError rather than being silently dropped, so a translated config stays close to what you intended. A small set is accepted but ignored because the native config has no equivalent — client.id and enable.auto.offset.store on the consumer, and sasl.oauthbearer.config (only valid alongside oauth_cb). If you set enable.auto.offset.store, note the shim manages offset positions itself (see Offsets and assignment) and does not honor that flag.

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()

produce(topic, value, key=, callback=/on_delivery=) queues a record. Delivery callbacks fire on poll() and flush(), as in confluent. value/key accept str (encoded UTF-8) or bytes. list_topics() returns cluster metadata (see below).

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())   # .topic() / .partition() / .offset() / .key() also available
    c.commit(msg)
c.close()

poll(timeout) returns one Message (or None), buffering the native batch underneath, so a confluent poll loop works unchanged. Always pass a timeout: poll() with no argument is non-blocking here, where confluent blocks (see below).

Batch consume

consume(num_messages, timeout) hands back a list of messages instead of one at a time, matching confluent’s batch API:

for msg in c.consume(num_messages=500, timeout=1.0):
    if msg.error():
        continue
    handle(msg.value())

Offsets and assignment

The offset and partition methods take and return confluent TopicPartitions (topic, partition, offset):

# commit specific offsets (absolute next-read positions, as in confluent)
c.commit(offsets=[TopicPartition("orders", 0, 1234)])

# where am I, where is the committed cursor
tps = c.assignment()                 # list[TopicPartition] currently assigned
positions = c.position(tps)          # .offset filled with the live consume position
committed = c.committed(tps)         # .offset = broker-committed cursor, or OFFSET_INVALID

c.unsubscribe()

commit() follows confluent’s three forms: commit(msg) commits that message’s offset + 1, commit(offsets=[...]) commits the given positions, and a bare commit() commits the last message handed to you (not the read-ahead batch position, so a crash never skips buffered-but-unhandled records).

Cluster metadata

md = c.list_topics()          # or p.list_topics() on a producer
for name in md.topics:        # md.topics: {name: TopicMetadata}
    ...

What differs from librdkafka

This is a behavioral shim, not a reimplementation:

• poll() / consume() need an explicit timeout. With no timeout argument they return immediately (non-blocking), where confluent blocks indefinitely. Pass poll(1.0) / consume(n, 1.0) as confluent examples do.
• A poll error from consume() raises KafkaException, where poll() returns it as an error Message (test if msg.error()). Both match confluent’s respective APIs.
• msg.error() returns a KafkaException, not confluent’s KafkaError. Truthiness (if msg.error(): ...) and str() work the same; code that introspects err.code() does not.
• commit(asynchronous=...) is always synchronous. The flag is accepted for signature compatibility but ignored; the commit completes before commit() returns.
• Delivery Message has no offset. The native single-produce path reports delivery success without a per-message offset, so the delivery callback’s msg echoes the produced topic/key/value but offset()/partition() are None. Consumed messages carry the real offset and partition.
• poll(timeout) on the producer may block until already-produced messages are acknowledged rather than honoring the timeout precisely.
• headers= and partition= on produce are accepted for signature compatibility but not yet forwarded.
• Manual assignment not yet shimmed. assign/seek/pause/resume are on the roadmap; the native KafkaConsumer exposes them today.
• ClusterMetadata carries topic names only; per-partition broker/replica detail is not yet populated.
• Unrecognized config keys raise a ValueError. See Configuration for the native knobs; anything outside the translated set is rejected loudly, except the small accepted-but-ignored set noted at the top (client.id, enable.auto.offset.store, sasl.oauthbearer.config).

For new code, prefer the native KafkaProducer / KafkaConsumer, which expose the batch-first, Arrow, and exactly-once APIs the shim hides.