A Common Lisp client library for Apache Kafka.
The public API is split between two packages:
-
cl-rdkafka/low-level
Nicknamed
cl-rdkafka/ll
, this package provides CFFI bindings for librdkafka. -
cl-rdkafka
Nicknamed
kf
, this package provides a higher-level interface 💅 with amenities such as garbage-collection ♻️, out-of-band error processing ↩️, and more!
Documentation for cl-rdkafka/ll
can be found in
librdkafka/rdkafka.h,
and kf
is documented under the API section.
(ql:quickload '(cl-rdkafka babel))
(let ((producer (make-instance
'kf:producer
:conf '("bootstrap.servers" "127.0.0.1:9092")
:serde #'babel:string-to-octets))
(messages '(("key-1" "value-1")
("key-2" "value-2"))))
(loop
for (k v) in messages
do (kf:send producer "topic-name" v :key k))
(kf:flush producer))
(ql:quickload '(cl-rdkafka babel))
(let ((consumer (make-instance
'kf:consumer
:conf '("bootstrap.servers" "127.0.0.1:9092"
"group.id" "consumer-group-id"
"enable.auto.commit" "false"
"auto.offset.reset" "earliest"
"offset.store.method" "broker"
"enable.partition.eof" "false")
:serde #'babel:octets-to-string)))
(kf:subscribe consumer "topic-name")
(loop
for message = (kf:poll consumer 2000)
while message
for key = (kf:key message)
for value = (kf:value message)
collect (list key value)
do (kf:commit consumer)))
;; => (("key-1" "message-1") ("key-2" "message-2"))
PRs and GitHub issues are always welcome and feel free to email me with any questions 📨
To run the tests:
$ docker-compose -f ./test/docker-compose.test.yml \
> up --build --remove-orphans --abort-on-container-exit test
$ docker-compose -f ./test/docker-compose.test.yml down --rmi all
$ docker system prune -fa && docker volume prune -f
To spin up and teardown a dockerized Kafka cluster to hack against:
# start a cluster on 127.0.0.1:9092
$ docker-compose up --build --remove-orphans -d
# tear the cluster down
$ docker-compose down --rmi all
# clean up after yourself
$ docker system prune -fa && docker volume prune -f
A client that produces messages to kafka topics.
make-instance
accepts the following keyword args:
-
conf
A required plist, alist, or hash-table mapping config keys to their respective values; both keys and values should be strings. The provided key-value pairs are passed as-is to librdkafka, so consult the librdkafka config docs for more info.
-
serde
An optional unary function accepting an object and returning a byte sequence; defaults to
#'identity
. -
key-serde
An optional unary function used to serialize message keys; defaults to
serde
. -
value-serde
An optional unary function used to serialize message values; defaults to
serde
.
Example:
(let ((producer (make-instance
'kf:producer
:conf '("bootstrap.servers" "127.0.0.1:9092")
:serde #'babel:string-to-octets))
(messages '(("key-1" "value-1")
("key-2" "value-2"))))
(loop
for (k v) in messages
do (kf:send producer "topic-name" v :key k))
(kf:flush producer))
((producer producer) (topic string) value &key key partition headers timestamp)
Asynchronously send a message and return a message
future
.
If partition
is not specified, one is chosen using the topic
's
partitioner function.
If specified, headers
should be an alist mapping strings to
byte-vectors.
timestamp
is the number of milliseconds since the UTC epoch. If not
specified, one will be generated by this call.
May signal partition-error
or condition from producer
's serde. A
store-function
restart will be provided if it's a serde condition.
((producer producer))
Block while in-flight messages are sent to kafka cluster.
A client that consumes messages from kafka topics.
make-instance
accepts the following keyword args:
-
conf
A required plist, alist, or hash-table mapping config keys to their respective values; both keys and values should be strings. The provided key-value pairs are passed as-is to librdkafka, so consult the librdkafka config docs for more info.
-
serde
An optional unary function accepting a byte vector and returning a deserialized value; defaults to
#'identity
. -
key-serde
An optional unary function used to deserialize message keys; defaults to
serde
. -
value-serde
An optional unary function used to deserialize message values; defaults to
serde
.
Example:
(let ((consumer (make-instance
'kf:consumer
:conf '("bootstrap.servers" "127.0.0.1:9092"
"group.id" "consumer-group-id"
"enable.auto.commit" "false"
"auto.offset.reset" "earliest"
"offset.store.method" "broker"
"enable.partition.eof" "false")
:serde #'babel:octets-to-string)))
(kf:subscribe consumer "topic-name")
(loop
for message = (kf:poll consumer 2000)
while message
for key = (kf:key message)
for value = (kf:value message)
collect (list key value)
do (kf:commit consumer)))
((consumer consumer) (timeout-ms integer))
Block for up to timeout-ms
milliseconds and return a message
or nil.
May signal partition-error
or condition from consumer
's serde. A
store-function
restart will be provided if it's a serde condition.
((consumer consumer) (topics sequence))
Subscribe consumer
to topics
.
Any topic prefixed with ^
will be regex-matched with the cluster's topics.
((consumer consumer) (topic string))
Subscribe consumer
to topic
.
If topic
starts with ^
, then it will be regex-matched with the cluster's
topics.
((consumer consumer))
Unsubscribe consumer
from its current topic subscription.
((consumer consumer))
Return a list of topic names that consumer
is subscribed to.
((consumer consumer) (partitions sequence))
Assign partitions
to consumer
.
partitions
should be a sequence of (topic . partition)
cons cells.
((consumer consumer))
Return a (topic . partition)
list of partitions assigned to consumer
.
((consumer consumer) &key offsets asyncp)
Commit offsets
to broker.
If offsets
is nil, then the current assignment is committed;
otherwise, offsets
should be an alist mapping (topic . partition)
cons
cells to either (offset . metadata)
cons cells or lone offset values.
On success, an alist of committed offsets is returned, mapping
(topic . partition)
to (offset . metadata)
.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: Same format as successful return valuebaddies
: An alist mapping(topic . partition)
tordkafka-error
If asyncp
is true, then a future
will be returned instead.
((consumer consumer) (partitions sequence) (timeout-ms integer))
Block for up to timeout-ms
milliseconds and return committed offsets
for partitions
.
partitions
should be a sequence of (topic . partition)
cons cells.
On success, an alist of committed offsets is returned, mapping
(topic . partition)
to (offset . metadata)
.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: Same format as successful return valuebaddies
: An alist mapping(topic . partition)
tordkafka-error
((consumer consumer) (partitions sequence))
Pause consumption from partitions
.
partitions
should be a sequence of (topic . partition)
cons cells.
partitions
is returned on success.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: A list of(topic . partition)
cons cellsbaddies
: An alist mapping(topic . partition)
tordkafka-error
((consumer consumer) (partitions sequence))
Resume consumption from partitions
.
partitions
should be a sequence of (topic . partition)
cons cells.
partitions
is returned on success.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: A list of(topic . partition)
cons cellsbaddies
: An alist mapping(topic . partition)
tordkafka-error
((consumer consumer))
Return consumer
's broker-assigned group member-id.
((consumer consumer) (timestamps list) (timeout-ms integer))
Look up the offsets for the given partitions by timestamp.
The returned offset for each partition is the earliest offset whose
timestamp is greater than or equal to the given timestamp in timestamps
.
timestamps
should be an alist mapping (topic . partition)
cons cells
to timestamp values.
On success, an alist of offsets is returned, mapping (topic . partition)
cons cells to offset values.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: Same format as successful return valuebaddies
: An alist mapping(topic . partition)
tordkafka-error
((consumer consumer) (topic string) (partition integer) (timeout-ms integer))
Query broker for low (oldest/beginning) and high (newest/end) offsets.
A (low . high)
cons cell is returned.
((consumer consumer) (partitions sequence))
Retrieve current positions (offsets) for partitions
.
partitions
should be a sequence of (topic . partition)
cons cells.
On success, an alist of positions is returned, mapping
(topic . partition)
to one of either:
- 1 plus the last consumed message offset
- nil if there was no previous message.
On failure, either an rdkafka-error
or partial-error
is signalled.
The partial-error
will have the slots:
goodies
: Same format as successful return valuebaddies
: An alist mapping(topic . partition)
tordkafka-error
((consumer consumer))
Close consumer
after revoking assignment, committing offsets,
and leaving group.
consumer
will be closed during garbage collection if it's still open;
this method is provided if closing needs to occur at a well-defined
time.
A kafka message as returned by consumer
's poll
or producer
's send
.
make-instance
should not be called with this class.
Example:
(let ((message (kf:poll consumer 5000)))
(kf:key message)
;; => "key-1", #(107 101 121 45 49)
(kf:value message)
;; => "Hello", #(72 101 108 108 111)
(kf:topic message)
;; => "foobar"
(kf:partition message)
;; => 0
(kf:offset message)
;; => 0
(kf:timestamp message)
;; => 1577002478269, :CREATE-TIME
(kf:headers message)
;; => '(("one" . #(1 2 3))
;; ("two" . #(4 5 6)))
)
((message message))
Return (values deserialized-key serialized-key)
from message
.
((message message))
Return (values deserialized-value serialized-value)
from message
.
((message message))
Return the topic message
originated from.
((message message))
Return the partition message
originated from.
((message message))
Return the offset for message
.
((message message))
Return (values timestamp timestamp-type)
from message
.
If timestamp is not available, then nil is returned. Otherwise:
timestamp
is the number of milliseconds since the UTC epochtimestamp-type
is either:create-time
or:log-append-time
((message message))
Return headers from message
as an alist mapping strings to byte vectors.
A future to hold the result of an async operation.
make-instance
should not be called with this class.
Example:
(let ((future (kf:send producer "topic" "message")))
(kf:donep future) ;; => nil
(kf:value future) ;; => #<MESSAGE {1005BE9D23}>
(kf:donep future) ;; => t
(let ((new-future (kf:then future
(lambda (message err)
(when err
(error err))
(kf:value message)))))
(kf:value new-future))) ;; => "message"
((future future))
Wait until future
is done and return its value or signal its condition.
((future future) (callback function))
Return a new future
that calls callback
when current future completes.
callback
should be a binary function accepting the positional args:
value
: the value that the current future evaluates to, or nil when it signals a condition.condition
: the condition signalled by the current future, or nil when it does not signal a condition.
callback
is called in a background thread.
((future future))
Determine if future
is done processing.
The conditions are structured in the following class hierarchy:
cl:serious-condition
cl:storage-condition
allocation-error
cl:error
kafka-error
rdkafka-error
partition-error
partial-error
Generic condition signalled by cl-rdkafka for expected errors.
Slot readers:
description
: Hopefully some descriptive description describing the error.
Condition signalled for librdkafka errors.
Slot readers:
enum
:cl-rdkafka/ll:rd-kafka-resp-err
enum symbol.description
:enum
description (inherited)
Condition signalled for errors specific to a topic's partition.
Slot readers:
topic
: Topic namepartition
: Topic partitionenum
:cl-rdkafka/ll:rd-kafka-resp-err
enum symbol (inherited)description
:enum
description (inherited)
Condition signalled for operations that partially failed.
Slot readers:
goodies
: Successful resultsbaddies
: Unsuccessful resultsdescription
:baddies
description (inherited)
Condition signalled when librdkafka functions fail to allocate pointers.
Slot readers:
name
: Name of the object that failed to be allocated.description
: Details about why the allocation may have failed.
The admin API is still baking 🍞, so it's not publicly exposed. The admin functionality is accessible if needed (see tests for usage examples), but it will be changing significantly in the near future.