// STRYKE-KAFKA — APACHE KAFKA CLIENT FOR STRYKE // PRODUCER + CONSUMER + ADMIN + CONSUMER-GROUP LAG

stryke package · cdylib libstryke_kafka.{dylib,so} · dlopen'd in-process on use Kafka · opt-in (kept out of stryke core)

Report GitHub Issues
// Color scheme

>_STRYKE-KAFKA

Streams without the JVM. Apache Kafka client for stryke — producer (keys, headers, partitions, binary payloads), consumer (offset commit, headers), consumer-group lag / watermarks / time-based offsets, and topic / cluster / config admin. Opt-in package backed by rdkafka (librdkafka bindings); ships as a Rust cdylib that stryke dlopens in-process on first use Kafka. The cdylib statically links librdkafka via rdkafka's cmake-build feature, so it is portable across glibc / musl / macOS without a system librdkafka.so.

Contents

Install

# from a release (no rustc + librdkafka build on the consumer machine)
s pkg install -g github.com/MenkeTechnologies/stryke-kafka

# from a local checkout
cd ~/projects/stryke-kafka
cargo build --release            # first build compiles librdkafka via cmake (~3-5 min); produces libstryke_kafka.{dylib,so}
s pkg install -g .               # cdylib lands in ~/.stryke/store/kafka@<version>/

# one-liner
make install

The cdylib is dlopened in-process on first use Kafka — no helper-binary fork per call. A shared tokio runtime + FutureProducer cache per brokers tuple is held in OnceCell, so producer batching/compression works as designed. Any stryke script that declares use Kafka resolves the package automatically.

Quick start: use Kafka

set brokers$ENV{KAFKA_BROKERS} = "localhost:9092"
ping (metadata fetch)p Kafka::ping() ? "alive" : "down"
produceval $r = Kafka::produce "events", "hello stryke", key => "k1"
bulk produceKafka::produce_many [{ value => "a", key => "1" }], topic => "events"
snapshot consumeval @msgs = Kafka::consume "events", group => "stryke-demo", limit => 100, timeout_ms => 5_000
callback per msgKafka::consume_stream "events", callback => fn { p "$_->{partition}.$_->{offset}: $_->{value}" }
list topicsp $_ for Kafka::topics()
create a topicKafka::create_topic "new-topic", partitions => 6, replication => 1
describe a topicval $info = Kafka::describe "new-topic"
consumer-group lagval $l = Kafka::lag "my-group", topic => "events"

A fuller end-to-end script — ping, produce, bulk produce, snapshot consume, list/create/describe topics:

use Kafka

$ENV{KAFKA_BROKERS} = "localhost:9092"

# Ping (metadata fetch).
p Kafka::ping() ? "alive" : "down"

# Produce one record.
val $r = Kafka::produce "events", "hello stryke", key => "k1"
p "wrote $r->{topic}/$r->{partition}@$r->{offset}"

# Bulk produce — array of { value, key? } hashrefs; topic is a named arg.
Kafka::produce_many [
    { value => "a", key => "1" },
    { value => "b", key => "2" },
    { value => "c", key => "3" },
], topic => "events"

# Snapshot consume — up to `limit` messages within `timeout_ms`.
val @msgs = Kafka::consume "events",
    group      => "stryke-demo",
    limit      => 100,
    timeout_ms => 5_000
@msgs |> ep

# Admin / metadata.
p $_ for Kafka::topics()
Kafka::create_topic "new-topic", partitions => 6, replication => 1
val $info = Kafka::describe "new-topic"
p to_json $info

Kafka::lag $group, topic => "..." reports per-partition lag (committed offset vs. high watermark). Kafka::watermarks and Kafka::offsets_for_times cover raw offset introspection. The full reference is below and in the README "API reference" section.

Configuration & brokers

Brokers can be passed on every call or set once via the KAFKA_BROKERS env var. The default is localhost:9092.

$ENV{KAFKA_BROKERS} = "localhost:9092"          # default

# per-call override (named arg on any broker op)
Kafka::topics(brokers => "kafka-1:9094,kafka-2:9094") |> ep

brokers is the only per-call connection option in v0.2.x — SASL / SSL connection options are deferred. Payload framing is controlled per call by the encoding option (utf8 default, plus base64 and hex for binary keys/values), accepted by produce, produce_many, and consume.

API — Producer

Kafka::produce$topic, $value, %opts → { topic, partition, offset }
Kafka::produce_many\@rows, %opts → $sent — requires topic => $name

produce opts: key, partition, timestamp (epoch millis), headers => { k => v }, encoding (utf8 default, base64, hex). produce_many rows are { value, key?, partition?, headers?, encoding? } hashrefs; the target topic is the required topic => "..." named arg, and an encoding opt sets the default for all rows. produce_many returns the count of records sent.

API — Consumer

Kafka::consume$topic, %opts → @messages
Kafka::consume_stream$topic, %opts → $count — fires callback per message

Opts: group, limit (default 10), timeout_ms (default 5000), encoding (utf8/base64/hex), commit (commit offsets after the drain — needs group). Snapshot-style: subscribe, poll up to limit messages or until timeout_ms runs out. consume_stream pulls the same snapshot and fires callback per message, returning the message count.

Message shape:

{
  topic, partition, offset, timestamp,
  key,                         # framed per `encoding` (null if missing)
  value,                       # framed per `encoding` (utf8 lossy by default)
  headers,                     # { k => v } object
}

API — Consumer offsets

Kafka::lag$group, %opts → { group, topic, partitions, total_lag } — opts: topic (required)
Kafka::watermarks$topic, %opts → { topic, partitions, total } — partitions: [{partition, low, high, count}]
Kafka::offsets_for_times$topic, $ts_ms, %opts → { topic, timestamp, partitions }
Kafka::committed$group, %opts → { group, topic, partitions } — opts: topic (required), partition (optional)

API — Admin

Kafka::topics%opts → @names
Kafka::describe$topic, %opts → { topic, partition_count, partitions } — partitions: [{id, leader, replicas, isr}]
Kafka::groups%opts → @groups — [{name, state, ...}]
Kafka::describe_group$group, %opts → { name, state, protocol, protocol_type, member_count, members }
Kafka::cluster%opts → { brokers, topic_count } — brokers: [{id, host, port}]
Kafka::create_topic$name, %opts → { created, errors } — opts: partitions, replication, config
Kafka::delete_topic$name, %opts → { deleted, errors }
Kafka::create_partitions$name, $partitions, %opts → { topic, altered, errors }
Kafka::describe_configs%opts → { resources, errors } — resource_type, resource_name
Kafka::alter_configs%opts → { altered, errors } — + entries => { k => v }
Kafka::delete_groups\@groups, %opts → { deleted, errors }
Kafka::delete_records$topic, $partition, $before, %opts → { topic, partition, results } — truncate a partition to $before (offset or latest)
Kafka::ping%opts → 1 | 0

API — Plumbing

Kafka::version() → $version_string — cdylib's CARGO_PKG_VERSION

API — Pure helpers (no broker)

Offline functions that compute Kafka semantics client-side — no broker round trip. Useful for partition prediction, rebalance planning, config validation, and topic-name hygiene.

Kafka::valid_topic_name$name → { name, valid, reason } — 1-249 chars of [a-zA-Z0-9._-], not . / ..
Kafka::sanitize_topic_name$name, $replacement? → { name, sanitized, changed } — coerce input into a valid topic name (illegal → _, truncate 249, reserved/empty → replacement)
Kafka::is_internal_topic$name → 1 | "" — __ prefix
Kafka::topics_collide$a, $b → 1 | "" — metric-namespace collision: equal after .→_
Kafka::parse_brokers$str → @{ {host, port} } — bootstrap.servers list
Kafka::build_brokers\@brokers → $str — inverse of parse_brokers
Kafka::normalize_brokers$str, %opts → $str — fill default :9092 (opt default_port), trim, dedupe
Kafka::partition_for_key$key, $partitions → { partition, hash } — JVM default partitioner: toPositive(murmur2(key)) % partitions
Kafka::partition_for_key_crc32$key, $partitions → { partition, crc32 } — librdkafka consistent: crc32(key) % partitions
Kafka::partition_for_key_fnv1a$key, $partitions → { partition, fnv1a } — librdkafka fnv1a: fnv1a(key) % partitions
Kafka::group_coordinator_partition$group, $partitions=50 → { group, partition, hash, partitions } — __consumer_offsets partition: abs(groupId.hashCode()) % 50
Kafka::transaction_coordinator_partition$transactional_id, $partitions=50 → { transactional_id, partition, hash, partitions } — __transaction_state partition
Kafka::range_assignment$partitions, @consumers → { assignment, partitions, consumers } — default RangeAssignor
Kafka::roundrobin_assignment$partitions, @consumers → { assignment, partitions, consumers } — RoundRobinAssignor (partition p → member p%N)
Kafka::sticky_assignment$partitions, \@consumers, \%previous? → { assignment, partitions, consumers } — StickyAssignor (KIP-54)
Kafka::replica_assignment$partitions, $replication_factor, \@brokers, %opts → { assignment, partitions, replication_factor } — rack-unaware partition→broker placement; opts: start_index, start_partition
Kafka::assignment_by_broker\@assignment → { brokers } — invert a partition→broker assignment to a per-broker view; inverse of replica_assignment
Kafka::assignment_diff\%previous, \%current → { revoked, assigned, moved } — rebalance plan; revoked = cooperative give-up set
Kafka::format_offset$n|$name → { offset, name } — -1 ↔ latest, -2 ↔ earliest
Kafka::topic_metric_name$name → { name, metric_name, has_collision_chars } — metric-namespace mangling (. → _)
Kafka::parse_topic_partition$value → { topic, partition } — split topic-N / topic:N
Kafka::format_topic_partition$topic, $partition, $separator? → $str — build topic-N (separator default -)
Kafka::compression_codec$value → { value, valid, canonical } — validate compression.type (none|gzip|snappy|lz4|zstd|producer)
Kafka::cleanup_policy$value → { value, valid, canonical, policies } — validate cleanup.policy (delete|compact|both)
Kafka::acks_value$value → { value, acks, all } — canonicalize producer acks (0|1|all/-1); dies on other values
Kafka::parse_duration_ms$value → $ms — parse a duration (number or ms|s|m|h|d|w suffix) to millis; -1 passes through
Kafka::isr_health\@replicas, \@isr, $min_isr? → { replicas, isr, min_isr, under_replicated, under_min_isr, healthy } — $min_isr default 1
Kafka::offset_lag\@partitions → { partitions, total_lag } — offline lag from [{partition, committed, high}]
Kafka::security_protocol$value → { value, valid, canonical } — validate security.protocol (PLAINTEXT|SSL|SASL_PLAINTEXT|SASL_SSL); canonical uppercase
Kafka::auto_offset_reset$value → { value, valid, canonical } — validate auto.offset.reset (earliest|latest|none|by_duration:<dur>)
Kafka::isolation_level$value → { value, valid, canonical, read_committed } — validate isolation.level (read_uncommitted|read_committed)
Kafka::timestamp_type$value → { value, valid, canonical } — validate message.timestamp.type (CreateTime|LogAppendTime)
Kafka::humanize_duration_ms$ms → { ms, human, unit } — inverse of parse_duration_ms (1w, 3d, 90s); -1 → "infinite"
Kafka::partition_distribution\@keys, $partitions, %opts → { partitions, partitioner, total, counts, used, empty, max, min, spread } — offline key-skew tally (opt partitioner: murmur2|crc32|fnv1a)

partition_for_key is a faithful port of Kafka's Utils.murmur2 (seed 0x9747b28c) plus the default partitioner's toPositive(hash) % partitions, so it predicts a keyed record's partition offline — no broker round trip.

Why a package, not a builtin

librdkafka is the canonical native Kafka client. Bundled here as an opt-in stryke package so workflows that need it don't bloat the core binary.

The stryke side is a thin .stk wrapper that calls kafka__* FFI symbols on the cdylib; the heavy code lives in libstryke_kafka.{dylib,so}, which statically links librdkafka via rdkafka's cmake-build feature and is dlopened in-process on first use Kafka. Core stryke is never linked against this package's deps.

FFI layer

Each Kafka::* wrapper builds a JSON args dict and calls a sibling kafka__* symbol resolved out of libstryke_kafka.{dylib,so}. The cdylib is dlopened in-process on first use Kafka (via stryke's pkg::commands::try_load_ffi_for resolver hook); a shared tokio runtime + a FutureProducer + admin-client cache per brokers tuple is held in OnceCell — no fork-per-call, so producer batching/compression works as designed. Responses are JSON; errors come back as a {error} payload and the wrapper dies with Kafka::<op>: <reason>. The authoritative export list is [ffi].exports in stryke.toml.

Tests

cargo test                                          # compiles, no live calls
KAFKA_BROKERS=localhost:9092 s test t/              # live round-trip

The end-to-end suite creates a temp topic, produces single + bulk messages, consumes them back, checks the count, and deletes the topic. It skips cleanly when $KAFKA_BROKERS isn't set or the broker isn't reachable. A local KRaft-mode broker (no ZooKeeper) via bitnami/kafka is documented in the README for round-trip testing.

Dev workflow

make             # release build (first time: ~3-5 min for librdkafka)
make debug
make test
make install
make clean

Roadmap

Shipped: produce with keys/headers/partitions/timestamps + binary framing, consume with headers/binary/offset-commit, consumer-group lag, watermarks, time-based offset lookup, topic config on create, create_partitions, describe/alter configs, delete_groups, describe_group, committed offsets, and delete_records (truncate to offset).

OpenSASL / SSL connection options; long-running consumer daemon (callback streaming); consumer seek / position (needs a persistent consumer handle)
LaterSchema Registry (Avro / Protobuf) value modes; Streams DSL (joins, windowed agg); transactional producer (EOS)

Layout

stryke-kafka/
├── Cargo.toml             # cdylib crate manifest (crate-type = ["cdylib"], publish = false)
├── src/
│   └── lib.rs            # cdylib — kafka__* extern "C" exports
├── lib/                   # stryke .stk wrapper(s) — `use Kafka`
├── stryke.toml            # stryke package manifest ([ffi] exports table)
├── t/                     # zunit-style tests
├── examples/              # runnable .stk examples
├── Makefile               # `make install` builds + installs
└── docs/                  # this site (GitHub Pages)

Sibling packages

Part of the stryke package family. Browse the others via the MenkeTechnologiesMeta umbrella repo: