@@ -1,5 +1,13 @@

# Changelog

## [1.2.0] - 2022-12-30

### Added

* Add support for offsets to `assign` and `assignment` consumer methods.

[issue #67](https://github.com/SahilKang/cl-rdkafka/issues/67) from

[@Uthar](https://github.com/Uthar).

## [1.1.0] - 2020-07-04

### Added

@@ -1,4 +1,5 @@

Copyright (C) 2018-2020 Sahil Kang <sahil.kang@asilaycomputing.com>

Copyright 2022 Google LLC

All parts of cl-rdkafka are licensed under the GNU GPLv3 as detailed below.

@@ -156,6 +156,15 @@ with any questions :incoming_envelope:

To run the tests:

:warning: Some of the following commands below, such as `--rmi` and `prune`,

will remove all local docker images and volumes. If this may be a problem,

consult the

[docker compose](https://docs.docker.com/engine/reference/commandline/compose_down/),

[docker system](https://docs.docker.com/engine/reference/commandline/system_prune/),

and

[docker volume](https://docs.docker.com/engine/reference/commandline/volume_prune/)

docs.

```bash

$ docker-compose -f ./test/docker-compose.test.yml \

> up --build --remove-orphans --abort-on-container-exit test

@@ -626,17 +635,23 @@ Return a list of topic names that `consumer` is subscribed to.

Assign `partitions` to `consumer`.

`partitions` should be a sequence of `(topic . partition)` cons cells.

`partitions` should be a sequence of either:

* `(topic . partition)` cons cells

* `((topic . partition) . offset)` cons cells

---

### assignment

```lisp

((consumer consumer))

((consumer consumer) &key offsetsp)

```

Return a `(topic . partition)` list of partitions assigned to `consumer`.

Return a list of partitions assigned to `consumer`.

The elements of the returned list will be either:

* `(topic . partition)` cons cells if `offsetsp` is nil

* `((topic . partition) . offset)` cons cells otherwise

---

@@ -6,7 +6,7 @@ services:

- "2181:2181"

kafka:

image: wurstmeister/kafka

image: wurstmeister/kafka:2.12-2.4.0

ports:

- "9092:9092"

environment:

@@ -1,4 +1,5 @@

;;; Copyright (C) 2018-2020 Sahil Kang <sahil.kang@asilaycomputing.com>

;;; Copyright 2022 Google LLC

;;;

;;; This file is part of cl-rdkafka.

;;;

@@ -88,7 +89,7 @@ Example:

(defgeneric committed (consumer partitions timeout-ms))

(defgeneric assignment (consumer))

(defgeneric assignment (consumer &key offsetsp))

(defgeneric assign (consumer partitions))

@@ -342,13 +343,20 @@ If ASYNCP is true, then a FUTURE will be returned instead."

(error (make-rdkafka-error err)))

(cffi:mem-ref rd-list :pointer))))

(defmethod assignment ((consumer consumer))

"Return a (topic . partition) list of partitions assigned to CONSUMER."

(defmethod assignment ((consumer consumer) &key offsetsp)

"Return a list of partitions assigned to CONSUMER.

The elements of the returned list will be either:

* (topic . partition) cons cells if OFFSETSP is nil

* ((topic . partition) . offset) cons cells otherwise"

(with-slots (rd-kafka-consumer) consumer

(with-toppar-list toppar-list (%assignment rd-kafka-consumer)

(let (partitions)

(foreach-toppar toppar-list (topic partition)

(push (cons topic partition) partitions))

(if offsetsp

(foreach-toppar toppar-list (topic partition offset)

(push (cons (cons topic partition) offset) partitions))

(foreach-toppar toppar-list (topic partition)

(push (cons topic partition) partitions)))

(nreverse partitions)))))

(defmethod committed

@@ -396,11 +404,29 @@ The PARTIAL-ERROR will have the slots:

(defmethod assign ((consumer consumer) (partitions sequence))

"Assign PARTITIONS to CONSUMER.

PARTITIONS should be a sequence of (topic . partition) cons cells."

PARTITIONS should be a sequence of either:

* (topic . partition) cons cells

* ((topic . partition) . offset) cons cells"

(with-slots (rd-kafka-consumer) consumer

(with-toppar-list

toppar-list

(alloc-toppar-list partitions :topic #'car :partition #'cdr)

(alloc-toppar-list

partitions

:topic (lambda (cons)

(let ((car (car cons)))

(if (consp car)

(car car)

car)))

:partition (lambda (cons)

(let ((car (car cons)))

(if (consp car)

(cdr car)

(cdr cons))))

:offset (lambda (cons)

(let ((car (car cons)))

(if (consp car)

(cdr cons)

cl-rdkafka/ll:rd-kafka-offset-invalid))))

(let ((err (cl-rdkafka/ll:rd-kafka-assign rd-kafka-consumer toppar-list)))

(unless (eq err 'cl-rdkafka/ll:rd-kafka-resp-err-no-error)

(error (make-rdkafka-error err)))))))

@@ -6,7 +6,7 @@ services:

- "2181:2181"

kafka:

image: wurstmeister/kafka

image: wurstmeister/kafka:2.12-2.4.0

ports:

- "9092:9092"

environment:

@@ -22,6 +22,6 @@ services:

dockerfile: ./test/Dockerfile.test

args:

bootstrap_servers: "kafka:9092"

librdkafka_commit: "v1.4.2"

librdkafka_commit: "v1.6.0"

links:

- kafka

@@ -1,4 +1,5 @@

;;; Copyright (C) 2018-2020 Sahil Kang <sahil.kang@asilaycomputing.com>

;;; Copyright 2022 Google LLC

;;;

;;; This file is part of cl-rdkafka.

;;;

@@ -181,6 +182,37 @@

(is (string= topic actual-topic))

(is (= partition actual-partition))))))

(test assign-offsets

(with-topics ((topic "consumer-assign-offsets-topic"))

(let ((consumer (make-instance

'kf:consumer

:conf (list "bootstrap.servers" *bootstrap-servers*

"group.id" "consumer-assign-offsets-group"

"enable.auto.commit" "true"

"auto.offset.reset" "earliest")))

(partition-1 35)

(partition-2 36)

(offset 3))

(kf:assign consumer (list (cons topic partition-1)

(cons (cons topic partition-2) offset)))

(destructuring-bind

(((actual-topic-1 . actual-partition-1) . actual-offset-1)

((actual-topic-2 . actual-partition-2) . actual-offset-2))

(kf:assignment consumer :offsetsp t)

(is (string= topic actual-topic-1))

(is (string= topic actual-topic-2))

(is (or (= partition-1 actual-partition-1)

(= partition-2 actual-partition-1)))

(if (= partition-1 actual-partition-1)

(progn

(is (= partition-2 actual-partition-2))

(is (= cl-rdkafka/ll:rd-kafka-offset-stored actual-offset-1))

(is (= offset actual-offset-2)))

(progn

(is (= partition-1 actual-partition-2))

(is (= offset actual-offset-1))

(is (= cl-rdkafka/ll:rd-kafka-offset-stored actual-offset-2))))))))

(test consumer-member-id

(with-topics ((topic "consumer-member-id"))

(let* ((group "consumer-member-id-group")

@@ -1,4 +1,5 @@

;;; Copyright (C) 2018-2020 Sahil Kang <sahil.kang@asilaycomputing.com>

;;; Copyright 2022 Google LLC

;;;

;;; This file is part of cl-rdkafka.

;;;

@@ -103,14 +104,14 @@

do (kf:send producer topic "deadlock-message")

finally (kf:flush producer))

(loop

repeat (* 2 expected-messages)

repeat (* 20 expected-messages)

until (= expected-messages actual-messages)

do (poll))))

(unwind-protect

;; pass or fail within 1 minute

;; pass or fail within 2 minutes

(loop

initially (lparallel:submit-task channel #'poll-loop)

repeat 30

repeat 60

until (= expected-messages actual-messages)

do (sleep 2)

finally (is (= expected-messages actual-messages)))

@@ -1 +1 @@

"1.1.0"

"1.2.0"