kafka-odin is a pure Odin Kafka client modelled on segmentio/kafka-go. The public surface mirrors kafka-go's Conn, Reader, Writer, and Dialer so a program familiar with the Go client feels at home in idiomatic Odin.

kafka-odin exists to give Odin programs a first-class Kafka client. The Go ecosystem already has kafka-go, a well-designed library that is easy to read, easy to test, and built directly on top of the Kafka wire protocol with no C dependencies. We follow the same shape in Odin: a low-level Conn that wraps a TCP socket and speaks the Kafka protocol, and higher-level Reader and Writer types that compose on top of it.

kafka-odin targets Kafka 2.7.0, 2.8.1, and 3.7.0, matching the test matrix of the upstream kafka-go library. Compact/flexible encodings introduced in the most recent versions of certain APIs are not yet implemented; every Kafka broker still accepts the older non-compact form supported here.

kafka-odin requires odin version dev-2026-04-nightly:a896fb2 or later.

The Conn type is the core of the kafka package. It wraps a TCP socket and speaks the Kafka wire protocol: each request is prefixed with its length and tagged with a correlation id that the broker echoes back on the response. A single mutex serializes round-trips, which is enough for synchronous use; multiplexed pipelining is deferred until the upstream mismatch-retry path is also ported.

import kafka "kafka-odin/kafka"

main :: proc() {
    conn, err := kafka.dial("localhost:9092")
    if err != .None do return
    defer kafka.conn_close(conn)

    versions, verr := kafka.conn_api_versions(conn)
    defer delete(versions)
    if verr != nil do return

    // versions[i].api_key, .min_version, .max_version
}

conn_round_trip(conn, api_key, version, body) is the workhorse: callers can encode any request via the per-API packages under protocol/ and ship it through that one entry point.

Reader consumes messages from a single (topic, partition) pair. Each reader_read_message call returns the next record from an internal RecordBatch buffer, fetching a fresh batch when the current one is exhausted.

r := kafka.reader_make(conn, "events", 0 /* partition */, 0 /* start_offset */)
defer kafka.reader_destroy(r)

for i in 0 ..< 100 {
    msg, err := kafka.reader_read_message(r)
    if err != nil do break
    fmt.printfln("offset=%d key=%s value=%s", msg.offset, msg.key, msg.value)
}

msg.key, msg.value, and msg.headers borrow from the Reader's internal buffer and remain valid only until the next reader_read_message call or reader_destroy. Callers that need to retain a message past those points must clone its bytes.

Consumer groups, broker-managed offset commits, and a background fetch loop are deferred to follow-up commits; the protocol packages they need (findcoordinator, joingroup, syncgroup, heartbeat, leavegroup, offsetcommit, offsetfetch) are already implemented and exposed as Conn-level helpers in kafka/group.odin.

Writer publishes messages to a single topic via a caller-supplied Conn. Each writer_write_messages call groups the supplied messages by partition through the configured Balancer, wraps each group as a v2 RecordBatch, and ships them in one Produce v3 request.

w := kafka.writer_make(conn, "events", 3 /* partition count */)
defer kafka.writer_destroy(w)

msgs := []kafka.Message {
    {key = transmute([]u8)string("a"), value = transmute([]u8)string("hello")},
    {key = transmute([]u8)string("b"), value = transmute([]u8)string("world")},
}
if err := kafka.writer_write_messages(w, msgs); err != nil {
    log.errorf("write failed: %v", err)
}

The MVP Writer is deliberately narrow: the caller manages the Conn lifecycle, supplies the partition count up front, and gets a synchronous round-trip per call. Batching across calls, async mode, automatic retries, and multi-broker leader discovery via Metadata are follow-up commits.

kafka-odin ships the four partitioner shapes upstream supports:

Each is wrapped through the generic Balancer type so they all plug into the Writer the same way.

Compression carries the wire-level codec id and codec_for(c) returns the matching Codec. The default state of each codec:

A Reader can already decode gzip'd topics produced by any other client. To wire the FFI codecs in, register your own Codec at startup:

import "kafka-odin/compress/snappy"
import kafka "kafka-odin/kafka"

kafka.codec_register(.Snappy, kafka.Codec {
    code = .Snappy,
    name = "snappy",
    compress   = proc(src: []u8, alloc: mem.Allocator) -> ([]u8, kafka.Codec_Error) {
        dst, ok := snappy.compress(src, alloc)
        return dst, ok ? .None : .Compression_Failed
    },
    decompress = proc(src: []u8, alloc: mem.Allocator) -> ([]u8, kafka.Codec_Error) {
        dst, ok := snappy.decompress(src, alloc)
        return dst, ok ? .None : .Decompression_Failed
    },
})

The same shape works for compress/lz4 and compress/zstd; each opt-in package adds its respective C-library dependency only when imported.

The SASL framework runs through a Mechanism vtable and the conn_sasl_authenticate driver. Three concrete mechanisms ship:

• Sasl_Plain for SASL/PLAIN. Transmits the password in the clear; deploy only over TLS-encrypted connections.
• Sasl_Scram for SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512. The salted-challenge-response exchange never transmits the password and is safe to use over PLAINTEXT listeners.

// PLAIN
plain := kafka.Sasl_Plain{username = "alice", password = "secret"}
m := kafka.sasl_plain_to_mechanism(&plain)
kafka.conn_sasl_authenticate(conn, &m)

// SCRAM-SHA-256
scram := kafka.sasl_scram_sha256("alice", "secret")
defer kafka.sasl_scram_destroy(&scram)
m := kafka.sasl_scram_to_mechanism(&scram)
kafka.conn_sasl_authenticate(conn, &m)

The kafka/tls opt-in package provides OpenSSL FFI bindings and a tls_dial_conn that returns a fully-wired ^kafka.Conn over an SSL-encrypted stream. Link with -lssl at build time; programs that do not import this package pay no OpenSSL dependency.

import "kafka-odin/kafka/tls"
import kafka "kafka-odin/kafka"

conn, err := tls.tls_dial_conn("broker.example.com:9093")
if err != .None do return
defer kafka.conn_close(conn)

// Every kafka.conn_round_trip / Writer / Reader call now travels through SSL.

Tls_Config covers the common options: skip_verify for development, ca_file for a custom PEM bundle, server_name for an explicit SNI override.

Five runnable programs under examples/ exercise the core APIs against a docker-compose broker:

docker compose up -d
odin run examples/api-versions
odin run examples/metadata -- events
odin run examples/producer -- localhost:9092 events 1
odin run examples/consumer -- localhost:9092 events 0
odin run examples/consumer-group -- localhost:9092 my-group events

kafka-odin/
├── kafka/                 # high-level types: Conn, Dialer, Reader, Writer, Transport, Balancer, etc.
├── protocol/              # one sub-package per Kafka API: encode_request / decode_response
│   ├── apiversions/
│   ├── metadata/
│   ├── produce/
│   ├── fetch/
│   ├── listoffsets/
│   ├── findcoordinator/
│   ├── heartbeat/
│   ├── joingroup/
│   ├── syncgroup/
│   ├── leavegroup/
│   ├── offsetcommit/
│   ├── offsetfetch/
│   ├── initproducerid/
│   ├── addpartitionstotxn/
│   ├── addoffsetstotxn/
│   ├── endtxn/
│   ├── txnoffsetcommit/
│   ├── createtopics/
│   ├── deletetopics/
│   ├── listgroups/
│   ├── describegroups/
│   ├── saslhandshake/
│   └── saslauthenticate/
├── examples/              # runnable usage examples
└── Makefile               # make check / make test / make docker

Unit tests cover every protocol package and the kafka package. They do not need a broker; make test runs the full suite locally in well under a second.

make check  # odin check -vet -strict-style -no-entry-point ./...
make test   # odin test ./... with the memory tracker enabled

CI runs two jobs on every push and pull request:

• unit: make test without a broker — this is the suite above.
• integration: a matrix over Kafka 2.7.0 / 2.8.1 / 3.7.0 that brings the broker up under docker compose, waits for it via scripts/wait-for-kafka.sh, and runs the integration-marked tests with KAFKA_INTEGRATION=1. Broker logs are captured on failure.