Mojo-kafka — Apache Kafka client for Mojo over librdkafka

General
1

Hey folks :waving_hand:

I’ve been pulling Kafka data into Python-based ML pipelines for years and have wanted to drop the Python hop on the hot path for a while. Mojo seemed like the natural place to do it — but there was no Kafka client. So I scaffolded one.

Repo: GitHub - dvirarad/mojo-kafka: Apache Kafka client for Mojo — librdkafka bindings with a Pythonic producer/consumer/admin API. Stream Kafka into your Mojo ML pipelines. · GitHub

It’s a thin wrapper over librdkafka (the C client behind almost every non-JVM Kafka library) exposed through Mojo’s FFI. The public API is Python-flavored on purpose so it’s familiar to anyone who’s used confluent-kafka-python:

⁠mojo
from kafka import Consumer, ConsumerConfig

fn main() raises:
var c = Consumer(ConsumerConfig(
bootstrap_servers=“localhost:9092”,
group_id=“my-app”,
auto_offset_reset=“earliest”,
))
c.subscribe([“events”])
while True:
var msg = c.poll(1000)
if msg:
run_inference(msg.value()) # straight into your Mojo model
c.close()
\

⁠ ### What’s in v0.1.0

  • Producer / Consumer / AdminClient with typed configs
  • Message with partition / offset / key / value (topic still TBD — issue #1)
  • Working examples including examples/ml_pipeline.mojo (Kafka → feature → toy inference loop)
  • CI: lint, smoke on Linux + macOS, integration against apache/kafka:3.7.0
  • Distributed via pixi — librdkafka and max pulled from conda automatically

Status

Alpha. The FFI layer compiles against real librdkafka symbols but rough edges remain — open as good-first-issues:

  1. Expose Message.topic
  2. Support headers (read + write)
  3. Typed KafkaErrorKind enum
  4. Transactional producer

Looking for feedback on

  • API shape — does Producer(cfg).produce(...) / Consumer(cfg).poll() feel idiomatic in Mojo?
  • The FFI layout in src/kafka/_ffi.mojo — is there a more idiomatic Mojo pattern for opaque C handles + struct field reads?
  • Whether anyone else has a streaming-Kafka-into-MAX use case I should hold space for in the API

PRs and issues very welcome. If you’ve wrapped C libraries in Mojo before, I’d love a sanity check on the OpaquePointer / UnsafePointer[T] lifetime story.

Thanks :fire: