kafka.q

Kafka from q: a client is a handle, a subscription delivers ONE table per batch to a target, a topic is a table coordinate, and the same handle publishes bytes or rows. hopen (`:pq:kafka:c1;cfg) opens the alias; cfg is librdkafka's configuration verbatim, a dict of its keys (metadata.broker.list, group.id, ...) to symbols or strings, judged at the open. The alias makes its consumer the first time it subscribes, commits, assigns or asks for its offsets, and its producer the first time it publishes or, with no client yet, asks for metadata; both come from cfg (a key librdkafka scopes to the other role is left out of each). .kafka.sub[h;`trades;`trade] decodes each batch's JSON into rows upserted into trade (types: explicit > trade's meta > a sniff); .kafka.sub[h;`trades;{[msgs] ...}] hands the raw batch table (topic partition offset msgtime msgkey headers data) to a function; either way the batch's offsets are committed once the target returns (or, with the option commit `manual, when it hands .kafka.commit what it processed), and a batch that fails is held unread, the topic paused and .kafka.errcb called - never a silent drop, never a redelivery loop. `:pq:kafka:c1:trades/ is the topic as an append-only table: select reads it to its high-water mark at the call, count get is 0N (the rows need a read: count select from it), meta and key decode a sample of its tail, and upsert publishes rows as JSON. h (`trades;msgkey;bytes) publishes one message and waits for its delivery; neg h is fire-and-forget. hclose closes the alias; .pq.conns[] lists it.

Examples

h:hopen (`:pq:kafka:c1;`metadata.broker.list`group.id!`localhost:9092`myapp)

.kafka.pub[h;`trades;"AAPL";"{\"sym\":\"AAPL\",\"px\":1.5}"]

`:pq:kafka:c1:trades/ upsert ([] sym:`AAPL`MSFT; px:1.5 2.5)

select from `:pq:kafka:c1:trades/ where partition=0, offset within 0 100

count select from `:pq:kafka:c1:trades/

get `:pq:kafka:c1:trades/

.kafka.read[h;`trades]

.kafka.read[h;(`trades;`partition`offset!(0;100 200))]

.kafka.sub[h;`trades;`trade]

.kafka.sub[h;`quotes;(`quote;`meta`store_rejects!(`topic`offset`msgtime;1b))]

.kafka.subs h

.kafka.subs[]

.kafka.errcb:{[h;code;reason] -2 "kafka ",string[code]," ",reason`error;}

Entity Summary

Entities	Short Description
.kafka.SAMPLE
.kafka.TIMEOUT	The ms a bounded read waits for a quiet partition before it signals; the messages per partition a sniff samples.
.kafka.assign[h;parts]	Assign partitions to the alias by hand instead of a group subscription: a dict of topic to partitions, a table of topic partition [offset], or `:: to clear the assignment.
.kafka.commit[h;x]	Commit what a target processed, synchronously: x is its rows (any table with topic partition offset - the meta option keeps them) or the misc of a batch from one partition.
.kafka.errcb	Called with every error an alias meets: a message that does not decode or a target that signals (code `decode or `target), a refused commit, librdkafka's own errors; reason is `topic`partition`offset`error!(...), the message fields null for a library error.
.kafka.flush[h;ms]	Wait up to ms for every message the alias has queued to be delivered.
.kafka.metadata[h]	The broker's topics as a table of topic partition leader replicas low high; a topic deleted while it is listed is left out.
.kafka.offsets[h]	The alias's assigned partitions: topic partition committed position low high lag.
.kafka.pub[h;topic;msgkey;data]	Publish to a topic: one message, or one per element when data is a list (key then `:: for none, one key for all, or a list of keys of the same count).
.kafka.read[h;spec]	Read a topic to its high-water mark at the call: every message, decoded as .kafka.sub decodes the topic (JSON when unsubscribed), message columns first.
.kafka.statcb	Called with librdkafka's statistics JSON (a string) every statistics.interval.ms, when that key is set.
.kafka.sub[h;topic;target]	Subscribe the alias to a topic.
.kafka.subs[h]	The subscriptions of an alias (h) or of every alias ([]): topic target decode messages rejected tombstones lastoffset paused, with the handle first when every alias is listed.
.kafka.unsub[h;topic]	Drop a topic from the alias's subscription.
.kafka.version[]	librdkafka's version, the first thing to check when a client does not open.

Entity Details

.kafka.assign[h;parts]

Assign partitions to the alias by hand instead of a group subscription: a dict of topic to partitions, a table of topic partition [offset], or `:: to clear the assignment.

Examples

.kafka.assign[h;(enlist `trades)!enlist 0 1]

.kafka.commit[h;x]

Commit what a target processed, synchronously: x is its rows (any table with topic partition offset - the meta option keeps them) or the misc of a batch from one partition. Each partition's commit is the offset after the last one x holds: the +1 is the library's, never yours. For a subscription with commit `manual, which never commits by itself; under `auto every batch commits as its target returns.

Parameters:	x - (any) the processed rows (a table), or a batch's misc (a dict)
throws:	kafka - the broker refused the commit: the partition is named type - x is neither, or a misc of several partitions (which cannot say where each one ends)

Examples

.kafka.commit[h;([] topic:enlist `trades; partition:enlist 0i; offset:enlist 0)]

.kafka.errcb

Called with every error an alias meets: a message that does not decode or a target that signals (code `decode or `target), a refused commit, librdkafka's own errors; reason is `topic`partition`offset`error!(...), the message fields null for a library error. The default writes one line to stderr; replace it as you would .z.pc - a reload of the library keeps the one you set.

.kafka.flush[h;ms]

Wait up to ms for every message the alias has queued to be delivered.

throws:	kafka Local: Timed out when some are still queued

Examples

.kafka.flush[h;5000]

.kafka.metadata[h]

The broker's topics as a table of topic partition leader replicas low high; a topic deleted while it is listed is left out.

Examples

.kafka.metadata h

.kafka.offsets[h]

The alias's assigned partitions: topic partition committed position low high lag. position is the next offset the alias reads (the delivered position, else the committed one, else low) and lag is high-position. The table is empty until the group assigns the partitions, in the first wakes after .kafka.sub; caught up is a non-empty table whose lag is all 0.

Examples

.kafka.offsets h

.kafka.pub[h;topic;msgkey;data]

Publish to a topic: one message, or one per element when data is a list (key then `:: for none, one key for all, or a list of keys of the same count). The messages are queued; .kafka.flush or a sync apply h (topic;msgkey;data) waits for delivery.

Parameters:	data - (any) a string or byte vector (ONE message), :: (a tombstone: Kafka's null value), or a list of them; the whole call is checked before anything is queued, and () is 'kafka (nothing to publish) msgkey - (any) a string, byte vector or symbol, or `:: for no key

Examples

.kafka.pub[h;`trades;"AAPL";"{\"sym\":\"AAPL\",\"px\":1.5}"]

.kafka.pub[h;`trades;`;("{\"sym\":\"IBM\",\"px\":1}";"{\"sym\":\"IBM\",\"px\":2}")]

.kafka.read[h;spec]

Read a topic to its high-water mark at the call: every message, decoded as .kafka.sub decodes the topic (JSON when unsubscribed), message columns first. (topic;opts) narrows or relaxes it. Three keys bound the read: partition (an int atom or list), offset (a from-to pair, inclusive as within; a null end is open) and msgtime (the earliest time); anything beyond them is a select. The rest are .kafka.sub's decode options under the same names: types, decode, ignore_errors, store_rejects, rejects_table, tombstones (default `row here) and path. A read is strict: a message that does not decode signals 'kafka naming it, unless ignore_errors skips it or store_rejects keeps it in the rejects table (keyed by topic partition offset). Any other key, or a bad value, signals 'kafka naming the option.

Examples

.kafka.read[h;`trades]

.kafka.read[h;(`trades;`partition`offset!(0;100 200))]

.kafka.read[h;(`trades;(enlist `ignore_errors)!enlist 1b)]

.kafka.statcb

Called with librdkafka's statistics JSON (a string) every statistics.interval.ms, when that key is set. The default does nothing; a reload of the library keeps the one you set.

.kafka.sub[h;topic;target]

Subscribe the alias to a topic. The target decides what a batch becomes: a symbol names a global table each batch's rows are upserted into (created when absent, upserted when keyed; its meta fixes the types), decoded from JSON by default; every target takes options as (target;opts); a function of three arguments {[rows;rejects;misc] ...} receives the decoded rows, the batch's reject records (under ignore_errors too) and misc (`topic`partitions`offsets`rows`tombstones, the last a table of the batch's tombstones); a function of ONE argument receives the raw batch table of topic partition offset msgtime msgkey headers data (msgkey and data byte vectors, headers a dict per message). The batch's offsets are committed when the target returns, unless the option commit is `manual. A message that fails to decode, or a target that signals, leaves the batch uncommitted, rewinds and pauses the topic, marks it paused in .kafka.subs and calls .kafka.errcb (code `decode or `target, the message named); subscribing again resumes it, and the held batch comes again; subscribing again to a topic that is not paused only changes its target, and unsub leaves a paused topic fetchable. The alias's config needs group.id. opts: types (a dict of column to type char, as .j.read takes; over the target's meta over a sniff of the batch); meta (the message columns to keep beside the decoded ones, from `topic`partition`offset`msgtime`msgkey, default none); decode (`json, `bytes or `string - the last two land one data column); ignore_errors (1b drops a message that fails to decode); store_rejects (1b keeps it in the rejects table, with topic partition offset msgtime msgkey error bytes - replayable, keyed by topic partition offset); rejects_table (symbol, default `reject_errors, implies store_rejects); max_per_wake (the messages one wake of the consumer delivers at most); path (handed to .j.read unchanged: the part of each message to land); tombstones (what a null value becomes: `row - typed nulls beside its message columns, `:: in data for a raw batch - `skip - counted in .kafka.subs - or `reject; a symbol target's default is `reject, every other target's `row); commit (`auto, the default: each batch commits as its target returns clean, and hclose or exit commits the last of them synchronously; `manual: nothing commits until the target - or anything later - hands .kafka.commit what it processed, and hclose or exit commits nothing more). Options are checked at the call: a bad one signals 'kafka naming it.

Parameters:	topic - (symbol) target - (any) a symbol, (symbol;opts), (function;opts) or a function
throws:	conn h is not an open alias

Examples

.kafka.sub[h;`trades;`trade]

.kafka.sub[h;`trades;(`trade;`types`meta!((enlist `px)!enlist "f";`offset`msgtime))]

.kafka.sub[h;`trades;{[rows;rejects;misc] `trade upsert rows}]

.kafka.sub[h;`trades;{[msgs] `raw upsert msgs}]

.kafka.sub[h;`trades;({[rows;rejects;misc] `trade upsert rows; .kafka.commit[h;rows]};`commit`meta!(`manual;`topic`partition`offset))]

.kafka.subs[h]

The subscriptions of an alias (h) or of every alias ([]): topic target decode messages rejected tombstones lastoffset paused, with the handle first when every alias is listed. decode is json, bytes or string, or batch for a one-argument target (it gets the raw batch); tombstones counts the null values the tombstones option `skip passed by.

Examples

.kafka.subs h

.kafka.unsub[h;topic]

Drop a topic from the alias's subscription.

Examples

.kafka.unsub[h;`trades]

.kafka.version[]

librdkafka's version, the first thing to check when a client does not open.

Return:	(string)
throws:	kafka librdkafka not found - set PEACHQ_KAFKA_LIB or install it

Examples

.kafka.version[]