moonkafka

An open-source Apache Kafka client driver written in MoonBit.

Target: Apache Kafka 4.x only.

This driver deliberately supports only the latest Kafka generation — KRaft-based clusters (no ZooKeeper), the modern Kafka protocol, and no legacy broker/version compatibility baggage.

Status: targeted for 0.2.0 — the transport + data-plane feature set is implemented and covered by a 200+ test suite (unit, mock-broker, golden fixtures) plus an optional real-cluster integration harness. The API may still change before 1.0.

Why MoonBit + Kafka?

• MoonBit compiles to small, fast WebAssembly (and native) targets, making it a good fit for lightweight producers/consumers in edge, serverless, and embedded environments.
• Kafka 4.x is a clean protocol baseline: by dropping ZooKeeper-era and pre-4.x compatibility, the driver stays small and easy to reason about.

Features

Working today:

• Kafka wire protocol codecs (compact types, zig-zag varints, tagged fields)
• Per-request API version negotiation against the broker's advertised ranges
• RecordBatch v2 (CRC32C-verified): record headers, control/transactional flags, read_committed filtering primitive, incremental batch builder, and whole-batch decompression for gzip, snappy, lz4, and zstd
• Data-plane APIs: Produce v12/v13 (topic-id addressing, per-record errors), Fetch v12-v16 with incremental fetch sessions (KIP-227), Metadata v12/v13, DescribeTopicPartitions v0 (paginated), ListOffsets v10/v11, FindCoordinator v4 (batched)
• Cluster layer: shared connection pool keyed by node id, metadata caching with expiry/error-triggered refresh, topic-id map, coordinator lookups
• Simple producer: per-leader connections, partitioner strategies (Kafka- compatible murmur2 key-partitioning; sticky batching or round-robin for keyless messages; per-send manual partition override), record batching with linger/batch-size/buffer-memory accounting, a background sender task (pipelined Produce requests per leader, acks 0/1/-1, retries with backoff bounded by delivery timeout), metadata refresh on leadership changes, REBOOTSTRAP_REQUIRED recovery
• Producer API: send, per-record SendHandles (await/cancel/ on_complete), batched send_all, and on-demand metrics (queue depth, in-flight, sent/failed counters, broker throttle time)
• Idempotent producer (acks=all default): InitProducerId handshake, per-partition sequence stamping with rewind on failure, epoch bump on UNKNOWN_PRODUCER_ID
• Transactions: transactional_id config with coordinator init, begin_transaction/commit_transaction/abort_transaction, AddPartitionsToTxn before produce, transactional offset commits (AddOffsetsToTxn + TxnOffsetCommit), EndTxn with epoch adoption, coordinator retry/refind, fencing detection, and abort-on-error commit policy
• Simple consumer: concurrent per-leader fetches with incremental sessions and eviction recovery, offset resolution by sentinel or timestamp, the seek family (explicit/timestamp/beginning/end), committed-offset tracking with sync/async commit and autocommit (interval + commit-on-close), max_poll_records / max_partition_fetch_bytes caps, per-partition pause/resume, auto-offset-reset policies, leader-epoch truncation detection, and read_committed filtering of aborted transactions
• KIP-848 consumer groups (primary path): ConsumerGroupHeartbeat membership with client-generated member ids, server-driven assignment applied atomically around rebalance listener hooks, static membership, regex subscription, graceful leave, and fencing recovery
• Consumer surface: subscribe/assign split, position/committed/ assignment/group_metadata introspection, max_poll_interval_ms enforcement, and utf8 record helpers
• Classic consumer groups (compat path): JoinGroup/SyncGroup/Heartbeat with range, round-robin, sticky, and cooperative-sticky assignors (two-round incremental rebalancing), static membership, graceful leave, and rebalance listener hooks
• group_protocol selection: KIP-848 (default), classic, or fallback ordering probed against the broker's advertised APIs
• Admin client: topics (create/delete/partitions/records/offset-delete), configs, ACLs, quotas, SCRAM credentials, log dirs, leaders, reassignments, cluster/controller introspection, transactions, and groups listing/describe/delete — with a shared retriable-result retry policy and paginated DescribeTopicPartitions walking
• Share groups (KIP-932): ShareConsumer over ShareGroupHeartbeat v1 membership, ShareFetch v2 acquisition with delivery-count caps, and ShareAcknowledge v2 (accept/release/reject/renew); Describe/Alter/ DeleteShareGroupOffsets admin ops
• Telemetry (KIP-714): GetTelemetrySubscriptions/PushTelemetry client driving a pluggable metrics provider (e.g. the driver's own counters)
• Pipelined broker connections: request timeouts, in-flight cap, reconnect through bootstrap servers, broker throttling
• TLS (including verified certificates via a custom CA) and SASL (PLAIN, SCRAM-SHA-256/512, OAUTHBEARER)

Requirements

• MoonBit toolchain (latest stable)
• An Apache Kafka 4.x cluster (KRaft mode). Older brokers are not supported and will not be.

Installation

moon add daqing/moonkafka

Quick start

///|
async fn main {
  // Produce
  let producer = @moonkafka.Producer::connect(
    host="127.0.0.1",
    port=9092,
    topic="events",
  )
  defer producer.close()
  let offset = producer.send(
    key=@utf8.encode("k1"),
    value=@utf8.encode("hello"),
  )
  println("produced at offset \{offset}")

  // Consume
  let consumer = @moonkafka.Consumer::connect(
    host="127.0.0.1",
    port=9092,
    topic="events",
    start_from=@moonkafka.StartFrom::Earliest,
  )
  defer consumer.close()
  for ;; {
    for record in consumer.poll() {
      println("offset=\{record.offset} value=\{record.value}")
    }
  }
}

A runnable example lives in cmd/main — it consumes all partitions of a topic and prints records until interrupted:

moon run cmd/main -- consume events [host] [port]

It can also produce a single message:

moon run cmd/main -- produce events "hello" [key] [host] [port]

Note: the socket layer is native-backend only (moonbitlang/async), so the module targets native.

Demo project

A standalone demo application lives in its own repository, daqing/moonkafka-demo. Clone it and run it:

git clone https://github.com/daqing/moonkafka-demo
cd moonkafka-demo

# start the bundled single-node Kafka 4.3 KRaft broker
docker compose -f docker-compose.kafka.yml up -d

# send one message, then read it back
moon run --target native cmd/main -- produce events "Hello from MoonBit" demo-key
moon run --target native cmd/main -- consume events

The project takes this library as an ordinary module dependency (daqing/moonkafka@0.2.1) and implements its own producer and consumer on top of the public Producer and Consumer APIs: its command-line layer parses arguments and renders results, while broker connection, metadata, the Kafka protocol, and message production and fetching all come from this library. Its README covers creating the events topic and the rest of the setup, and it carries a container-backed end-to-end integration test (make itest) that drives the real CLI against a real broker. Bilingual documentation: https://daqing.github.io/moonkafka-demo/.

The in-repository example in cmd/main covers the same client patterns in a single package — consume-group included.

Development

moon test           # unit + mock-broker (fake broker) suite
make fmt            # format code
make info           # regenerate package interfaces (.mbti)
make integration    # Kafka 4.3 (docker or podman) + real-client smoke test
make bench          # produce/consume throughput (needs a broker)
make docker-up      # start a KRaft cluster (make docker-up MULTI=1 for 3 nodes)

Full harness guide: docs/5-testing.md. Protocol notes: version negotiation, consumer groups, transactions.

License

MIT.

The third-party components this library links or redistributes — the MoonBit standard library and packages, zlib, and Zstandard — are listed with their licenses in THIRD_PARTY_NOTICES.md.