Crate llingr_kafka

Expand description

The llingr concurrent message processing engine with a Kafka broker, batteries included.

One pre-baked crate: the llingr-demux Go engine and its franz-go broker layer compile into your binary as a static c-archive during cargo build. The pure-Go client speaks Kafka and every Kafka-compatible broker, RedPanda and Amazon MSK included. Per-key concurrent processing without head-of-line blocking, contiguous offset commit guarantees, engine logs on the process-global log facade under the target LOG_TARGET, and baked-in Prometheus metrics activated at runtime with Metrics. There are no cargo features and no broker selection.

§Quick start

use llingr_kafka::{AutoOffsetReset, Builder, Message, Metrics, Options, Traits};
use llingr_kafka::{DeadLetterHandler, ProcessHandler};

struct MyProcessor;

impl ProcessHandler for MyProcessor {
    fn process(&self, msg: &Message) -> Result<Traits, Box<dyn std::error::Error>> {
        // Keys are frequently PII: log coordinates, never key contents.
        println!("partition={} offset={}", msg.partition(), msg.offset());
        Ok(Traits::none())
    }
}

// Required alongside the processor: failed messages need somewhere to go.
// Logging is the bare minimum; a real DLQ topic or table is recommended.
struct MyDeadLetters;

impl DeadLetterHandler for MyDeadLetters {
    fn handle(&self, msg: &Message, error_msg: &str) -> Result<(), Box<dyn std::error::Error>> {
        eprintln!("dead-letter partition={} offset={} reason={}", msg.partition(), msg.offset(), error_msg);
        Ok(())
    }
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = Builder::new("orders", MyProcessor, MyDeadLetters)
        .brokers("broker1:9092,broker2:9092")
        .consumer_group("orders-svc")
        .options(Options::new().auto_offset_reset(AutoOffsetReset::Earliest))
        .metrics(Metrics::serve("0.0.0.0:9464", "/metrics"))
        .build()?;

    let stop = engine.stopper(); // Send closure for signal-watcher threads
    engine.run()?; // BLOCKS until stop() or an emergency shutdown
    Ok(())
}

§Operational constraints

Hosting the Go runtime in-process has hard rules. Read these before shipping.

One engine per process. The Go runtime is a process-global singleton. A second Builder::build returns a clean error.
The Go runtime initialises at process start. The statically linked engine registers a loader initialiser: the Go runtime comes up before main() runs, not at build(), bringing its signal handlers for SIGSEGV, SIGBUS, SIGFPE, SIGPROF, and SIGURG with it. The engine never claims SIGINT or SIGTERM, so registering handlers for those works at any point; signal-hook, for example, chains to any pre-existing handler. Do not install non-chaining handlers for the fault signals, and anything touching them must use SA_ONSTACK.
Do NOT call stop() from a signal handler. Go code is never async-signal-safe: calling stop() or any engine function from inside a signal handler can deadlock or crash the process. Use signal_hook::flag to set an atomic flag, then poll it from a normal thread and call stop.
Panics. Rust panics in handlers are caught at the FFI boundary and converted to error codes; a panic in the process handler routes the message to the dead-letter handler with the reason “panic in process callback”. This protection relies on the host being built with panic = "unwind", the default. Under panic = "abort" the first handler panic aborts the process; the crate’s build script warns loudly when a profile does this. Go panics in engine internals abort the entire process with no recovery possible.
Thread budget. Handlers run on Go runtime threads, and a blocking handler pins one for its duration. GOMAXPROCS and GODEBUG must be set as environment variables before process start; setting them from Rust code has no effect. If you also run tokio or rayon, budget threads explicitly.
cargo test. All tests within a single test binary share one Go runtime instance. Use --test-threads=1 or a serialisation crate if tests depend on engine state.
musl/Alpine is not supported yet: an upstream Go limitation; the build script fails with the full explanation and links. Use a glibc image such as Debian or Ubuntu; a scratch image works because the binary is static.

§Exit and lifecycle hygiene

Borrowed message data is callback-scoped, and retaining it is worse than a use-after-free. The key/value slices point into a broker record buffer and a pooled engine work item, both recycled after the callback returns. A stored slice will later read a different message’s bytes: silent wrong data, not necessarily a crash. Copy out anything you need to keep with to_vec() or to_string().
Shutting down: call stop, don’t just exit. The call that initiates shutdown drains in-flight work, commits offsets, and returns from run. std::process::exit skips the drain; that is safe, because uncommitted offsets are redelivered at least once on restart, but wasteful. Never call stop() from inside a process or dead-letter handler; signal another thread.
emergency_stop abandons in-flight work: no drain, no final commit, and duplicates on restart are expected; the engine’s contract is at-least-once. The shutdown handler receives the reason exactly once, on graceful and emergency exits alike.
Liveness is your responsibility. If a handler stalls, nothing crashes: run() simply blocks. The engine’s own resilience covers the broker side: sustained partition poll failure triggers an emergency shutdown with the reason after a bail window, ten minutes by default.

§Architecture

Kafka / RedPanda / MSK
    |
Go engine, statically linked (libllingr.a)
    |  franz-go broker client -> llingr-demux pipeline
    |
C FFI boundary (ABI v1, checked at build())
    |
This crate: safe wrapper + llingr-nexus contracts
    |
Your application (ProcessHandler / DeadLetterHandler)

Re-exports§

pub use snapshot::Snapshot;

Modules§

snapshot: Typed view of the engine’s snapshot document.

Structs§

Builder: Staged construction of the engine (Llingr).
DemuxConfig: Engine settings, mirroring Go’s config.DemuxConfig. Anything left unset receives the engine’s production default, and every value is range-validated at build time, reported as a clean error rather than a crash. The defaults are good enough for most situations.
Header: One header: a UTF-8 key and an optional (nullable) byte value.
Headers: Ordered, borrowed view over a record’s headers.
Llingr: The llingr engine handle.
LlingrError: Error type for llingr operations.
Message: A complete record delivered to the process callback.
Metrics: Prometheus metrics activation, passed to the engine builder’s .metrics(...) hook.
MetricsHandle: Handle returned by Metrics::registry: renders the current OpenMetrics exposition for serving on the application’s own HTTP stack.
Options: Typed option builder for the Kafka client.
Traits: The 64-bit trait bit field.

Enums§

AutoOffsetReset: How the consumer starts when no committed offset exists for a partition (both adapters’ auto.offset.reset option).
BalanceStrategy: Consumer-group partition assignment strategy (partition.assignment.strategy).
ClientLogLevel: Verbosity of the Kafka client’s internal diagnostics (llingr.client.log.level), which the engine bridges into the log facade alongside its own lines.
Timestamp: A record’s timestamp and how it was set: one enum carries the instant and its kind (rdkafka’s shape), so there is no separate “type” field to juggle. Milliseconds since the Unix epoch (the Kafka wire resolution); the named millis field keeps the unit visible at match sites.

Constants§

LOG_TARGET: Log target used for all engine lines, so applications can filter or re-route them (e.g. RUST_LOG=llingr=debug).
OPENMETRICS_CONTENT_TYPE: The Content-Type for the OpenMetrics text exposition, matching what Go’s promhttp serves with EnableOpenMetrics and what the reference scrapers expect. Use it when mounting MetricsHandle::scrape output on your own HTTP stack.

Traits§

DeadLetterHandler: Required handler for messages that failed processing.
ProcessHandler: Required handler processing each message.
ShutdownHandler: Optional handler for consumer shutdown notification.