cosmonic:kafka

cosmonic:kafka@0.5.0 is the Kafka capability built into the Cosmonic Control host. A hostPlugins entry for it names no image or file, and there is no provider to deploy. For how to enable and configure the plugin on a host group, see the cosmonic:kafka host plugin. This page is the interface reference.

The WIT is fetched from the digest-pinned ghcr.io/cosmonic-labs/cosmonic/kafka package.

Bindings are capabilities

A guest never supplies a broker, credential, group, or arbitrary client property to a WIT function. The workload's resolved host-interface binding supplies all of them before the component runs.

That is why consumer.open() takes no arguments and why there is no producer resource to construct. A plain import uses the unnamed binding; a named import, written with (implements ...), selects a named binding and is the explicit way to grant another cluster, principal, group, or client policy.

A provider retains interface declarations per binding. Unioning every cosmonic:kafka interface across a component would let a named import reach an interface declared only on another binding.

Choosing an interface

Interface	Use it for	Shape
producer	Publishing	Binding-scoped functions, no resource
transaction	Atomic publishing, and offset commits alongside them	begin() returns a stateful resource
consumer	A long-lived service needing direct control of a Kafka session	A stateful resource your component owns
handler	Event-driven processing that scales to zero	Your component exports it; the host polls
types	Shared vocabulary	Records, variants, and the error-code variant

handler is the serverless path and consumer is the service path. With handler, the provider owns polling, heartbeats, assignment, retries, and commits, and invokes your component only when a partition has work. That means growing a component pool does not add Kafka group members. With consumer, your component owns group membership and must keep reading the record stream, because the heartbeat is driven by that stream: a consumer whose stream is not being read is eventually evicted from its group, the async equivalent of exceeding max.poll.interval.ms.

Binding configuration

Configuration resolves from config, configFrom and secretFrom under the host's normal layering rules. Broker addresses and non-secret policy go in config; credentials go in a Secret named by secretFrom. Unknown or malformed properties fail binding rather than being ignored, because silently dropping a durability, authentication or isolation property would change the contract.

Across all shapes, topics is the authorization ceiling for every topic operation on the binding.

Ordinary producer

hostInterfaces:
  - namespace: cosmonic
    package: kafka
    version: "0.5.0"
    interfaces: [producer]
    config:
      bootstrap.servers: kafka.svc:9092
      topics: orders.created,orders.updated
      acks: all
      compression.type: zstd
    secretFrom:
      - name: kafka-orders-credentials

The imported interface is the producer capability. There is no open, close or global flush. Each send reports its own delivery outcome, and the provider keeps the shared native client for the binding's lifetime.

Transactional producer

hostInterfaces:
  - name: orders-tx
    namespace: cosmonic
    package: kafka
    version: "0.5.0"
    interfaces: [transaction]
    config:
      bootstrap.servers: kafka.svc:9092
      topics: orders.input,orders.output
      transactional.id: order-processor-0
      transaction.group.id: order-processor

transactional.id is required and must stay stable across restarts for Kafka fencing and recovery. transaction.group.id is required only if the workload calls send-offsets.

A binding must not grant both producer and transaction over the same native producer: declare separate bindings when you need both. And every simultaneously live replica needs its own stable transactional ID; reusing one across hosts or replicas causes Kafka to fence one of them. Elastic transaction pools therefore need an external stable-ID allocation scheme, which this interface does not provide.

Host-driven handler

hostInterfaces:
  - namespace: cosmonic
    package: kafka
    version: "0.5.0"
    interfaces: [handler, producer]
    config:
      bootstrap.servers: kafka.svc:9092
      topics: orders.input,orders.output,orders.dlq
      handler.topics: orders.input
      handler.group.id: order-transformer
      handler.batch.size: "32"
      dead-letter.topic: orders.dlq

handler.topics is the subscription; topics is the ceiling. The two differ because a handler usually reads one topic and produces to another. handler.group.id is the provider-owned consumer group, handler.batch.size caps the records dispatched in one call, and dead-letter.topic receives permanent failures; it must itself sit inside the topics grant.

handler.topics is the gate, not one of the checks. A binding that omits it binds successfully and is silently never dispatched to, on the reasoning that a component exporting handler without asking for dispatch is a library world rather than a mistake. The only trace is a debug log. A typo there produces a green deploy that processes nothing, so check it first when a handler looks idle.

Once handler.topics is present, omitting bootstrap.servers, handler.group.id or dead-letter.topic does fail the bind, as does a transaction binding without transactional.id. A pull consumer without consumer.group.id stays valid for manual assignment, but its subscribe call fails.

Handler config is read only from the unnamed binding, and only one handler component per workload is permitted.

producer

Every function is async: a host component plugin installs concurrent host functions on the caller's linker, and a synchronous function cannot be served that way.

Function	Signature
send	(topic: string, %record: produce-record) -> result<produce-ack, error>
send-batch	(topic: string, records: list<produce-record>) -> result<list<result<produce-ack, error>>, error>
send-stream	(topic: string, records: stream<produce-record>) -> future<result<_, error>>
partition-count	(topic: string) -> result<u32, error>
watermark-offsets	(topic: string, partition: partition) -> result<watermarks, error>

send-batch results are positional: entry N is the outcome for record N. Records are enqueued before their delivery reports are awaited so the provider can batch them, and a record rejected before enqueue appears as an error in its own position without failing unrelated entries.

send-stream applies backpressure: the provider stops reading while its bounded queues are full. The returned future resolves only after the input ends and every accepted record has an outcome. If the workload, binding or provider stops first, the future reports an error, because accepted records may have an unknown outcome.

Common errors: msg-timed-out (not acknowledged within message.timeout.ms), queue-full (the binding's bounded local queue), unknown-topic-or-part (topic absent, auto-create off), not-leader-for-partition (leadership moved; retry after refresh).

transaction

begin() initializes the binding's transactional client (fencing any previous producer using the same transactional.id) and leases its single transaction slot. The initial contract permits one active transaction per binding; a concurrent begin fails with state or a retriable resource error.

Function	Signature
begin	() -> result<transaction, error>
partition-count	(topic: string) -> result<u32, error>
watermark-offsets	(topic: string, partition: partition) -> result<watermarks, error>

The transaction resource is a strict state machine. Producing and send-offsets are valid while active. A commit, abort or dropped resource ends it. Calls after a terminal operation fail with state. After an error carrying txn-requires-abort, the only valid terminal operation is abort. A fatal error retires the binding's native client.

Method	Signature
send	(topic: string, %record: produce-record) -> result<produce-ack, error>
send-batch	(topic: string, records: list<produce-record>) -> result<list<result<produce-ack, error>>, error>
send-stream	(topic: string, records: stream<produce-record>) -> future<result<_, error>>
send-offsets	(offsets: list<partition-offset>) -> result<_, error>
commit	() -> result<_, error>
abort	() -> result<_, error>

send-offsets takes the positions to resume at, normally one past the last record processed. The provider validates every topic and uses the binding's fixed transaction.group.id, so a guest cannot select or disturb another consumer group.

A concurrent commit or abort closes stream admission. Permits already granted may finish enqueueing, which defines the accepted prefix: commit waits for that prefix and includes it, abort waits until nothing can enqueue and discards. The stream future reports state when finalization truncates its input, so a caller cannot mistake the prefix for the complete stream.

consumer

A stateful resource holding group membership and its heartbeat. Every function here is async, as in producer and transaction.

open is the only static member in the package: open: static async func() -> result<consumer, error>. It takes no arguments because the binding fixes brokers, credentials, group identity, offset behavior, isolation level, assignment strategy, fetch policy and client limits. Note the provider forces enable.auto.offset.store=false and defaults enable.auto.commit=false, overriding an operator who sets them.

Subscription and assignment

Method	Signature
subscribe	(topics: list<string>) -> result<_, error>
unsubscribe	() -> result<_, error>
subscription	() -> result<list<string>, error>
assign	(partitions: list<partition-ref>) -> result<_, error>
incremental-assign	(partitions: list<partition-ref>) -> result<_, error>
incremental-unassign	(partitions: list<partition-ref>) -> result<_, error>
assignment	() -> result<list<partition-ref>, error>
assignment-lost	() -> bool
rebalance-protocol	() -> rebalance-protocol

subscribe replaces any previous subscription, and a topic name beginning with ^ is treated as a regular expression. Resolving does not mean partitions are assigned; that arrives on rebalances once the group settles. A provider rejects a regular expression when the binding has a finite literal topic grant, because comparing the expression text to that list cannot prove which topics it will match.

Reading

Method	Signature
records	() -> result<tuple<stream<consumed-record>, future<result<_, error>>>, error>
rebalances	() -> result<stream<rebalance-event>, error>
pause	(partitions: list<partition-ref>) -> result<_, error>
resume	(partitions: list<partition-ref>) -> result<_, error>

records and rebalances are each callable once. A second call returns state. And calling rebalances switches the consumer to manual assignment: the provider stops answering rebalances for you, so a guest that takes the stream and never calls assign or incremental-assign gets no partitions. Rebalance events are dropped with a warning if the guest is not draining them.

Offsets

Method	Signature
commit	(offsets: list<partition-offset>) -> result<list<partition-result>, error>
committed	(partitions: list<partition-ref>) -> result<list<partition-result>, error>
position	(partitions: list<partition-ref>) -> result<list<partition-result>, error>
seek	(partitions: list<partition-ref>, to: position) -> result<_, error>
offsets-for-times	(partitions: list<partition-ref>, time: timestamp-ms) -> result<list<partition-result>, error>
watermark-offsets	(topic: string, partition: partition) -> result<watermarks, error>
close	() -> result<_, error>

Rebalance events. rebalance-event is assign (partitions being added: accept with assign under the eager protocol or incremental-assign under cooperative), revoke (being taken away: commit progress first, because once acknowledged those offsets can no longer be committed), or lost (the assignment went without a clean revoke, typically after falling out of the group; another member may already own them).

rebalance-protocol is none (not in a group, or not yet joined), eager (stop-the-world: the whole assignment is revoked then reassigned), or cooperative (only the partitions that actually move are revoked).

handler

Your component exports this one. The provider imports it and calls in.

handle: async func(records: list<consumed-record>) -> result<option<offset>, handler-error>;

The batch is never empty, and the return value says how far it was handled:

• ok(none) — the whole batch. The provider commits past its last record.
• ok(some(offset)) — handled through offset inclusive and no further. The provider commits past that one and redelivers the rest, so a handler that processes seven of ten records keeps the seven instead of being given all ten again.
• error — none of it was handled.

An offset outside the batch is a caller error and is treated as error, since committing past records the provider never delivered would drop them.

handler-error is deliberately a variant rather than a types.error, and the distinction is the one thing the handler knows and the provider cannot infer:

• transient(option<string>) — try again. A dependency was unavailable, a lock was held, a quota was spent.
• permanent(option<string>) — this record will never succeed. On a single-record batch it is dead-lettered immediately. On a multi-record batch the whole batch is rewound and the head record redelivered alone for its own verdict, because a batch is only ever dead-lettered one record at a time. The batch ceiling also collapses to 1 after any failure and doubles back on success.

Core types

Type	Fields
consumed-record	topic, partition, offset, key, value, headers, timestamp, timestamp-type, leader-epoch
produce-record	partition, key, value, headers, timestamp
produce-ack	partition, offset, timestamp
partition-ref	topic, partition
partition-offset	topic, partition, offset, leader-epoch, metadata
watermarks	low, high
header	key, value

Aliases: timestamp-ms = s64, partition = s32, offset = s64.

error is a record, not a bare code:

Field	Type	Meaning
code	error-code	A large variant mirroring the underlying client's taxonomy, including msg-timed-out, queue-full, unknown-topic-or-part and not-leader-for-partition.
message	string	Human-readable detail. Not stable; do not parse it.
fatal	bool	The native client is unusable and must be rebuilt or reopened.
retriable	bool	The identical operation may be retried as-is.
txn-requires-abort	bool	The in-flight transaction must be aborted before further transactional work.

Branch on the three predicates, not on code. They are not derivable from it: the same code can be retriable in one context and fatal in another, so they are reported per occurrence.

partition-result, returned by commit, committed, position and offsets-for-times, carries topic, partition, offset (absent when that partition failed), leader-epoch, metadata, and error (absent on success).

position, the seek target, is beginning, end, stored, tail(u64) or exact(offset). timestamp-type is not-available, create-time or log-append-time.

Worlds

World	Imports	Exports
producer-client	types, producer	—
transactional-producer-client	types, transaction	—
consumer-service	types, consumer	wasi:cli/run@0.3.0
processor-service	types, consumer, transaction	wasi:cli/run@0.3.0
record-handler	types, producer	handler
host-plugin	host interfaces, types, handler	types, producer, transaction, consumer, workload lifecycle

consumer-service and processor-service export wasi:cli/run because they are called once and expected to run for the service lifetime. In processor-service, the consumer and transaction bindings must name the same consumer group when offsets and produced records need to commit atomically.

Delivery semantics

Handler delivery is at least once, and a trap, timeout, stopped workload, provider restart or uncertain handoff can redeliver records. Handlers must be idempotent.

Offsets usually move only after the handler reports progress, but not always. A permanent verdict on a single record is dead-lettered at once and the offset stored past it, and a record that only ever traps is rewound five consecutive times and then dead-lettered the same way. transient is deliberately excluded from that cap: it paces redeliveries with a 100 ms to 30 s backoff and retries indefinitely.

The provider never drops a record it has nowhere to put. dead-letter.topic is required at bind precisely so that case cannot arise, but if a record ever does fail with no dead-letter topic, the partition stops behind it and the provider escalates from a repeating warning to an error an operator can alert on.

Calls for different partitions may run concurrently; calls for one partition stay ordered and never overlap.

Clients and their queues are bounded host resources. Enforced limits include 64 clients per component per kind, 32 send-streams per component, 64 handler calls in flight, buffered-byte ceilings, and validation of native client configuration. There is no limit on the number of bindings.

send-stream also applies a per-record size limit of roughly 1 MiB; an oversized record comes back as an invalid-arg entry rather than failing the stream.