From bd4ec4afb3a4fc9e5f8b6c189b880685aaf79b29 Mon Sep 17 00:00:00 2001 From: Francois Date: Tue, 30 Oct 2018 07:28:49 +0000 Subject: [PATCH 1/7] minor: factored duplicated assertion, refined doc, fixed indentation --- README.md | 4 ++++ src/kinsky/client.clj | 56 +++++++++++++++++++------------------------ 2 files changed, 29 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 0e28112..22faca2 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,9 @@ Kinsky provides the following: **JSON**, **EDN** and a **keyword** serializer for keys. - Documentation +The `core.async` facade that used to be part of this project is now +[a separate project](https://github.com/fmjrey/kinsky-async). + ## Usage ```clojure @@ -50,6 +53,7 @@ Thanks a lot to these awesome contributors ### 0.1.25 - Removal of the asynchronous façade, transducers should suffice +- Minor typo and code refactoring ### 0.1.24 diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj index bfa1e77..34a06a2 100644 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -89,9 +89,9 @@ "Subscribe to a topic or list of topics. The topics argument can be: - - A simple string when subscribing to a single topic + - A simple string or keyword when subscribing to a single topic - A regex pattern to subscribe to matching topics - - A sequence of strings + - A collection of strings or keywords The optional listener argument is either a callback function or an implementation of @@ -417,9 +417,17 @@ :by-partition (group-by (juxt :topic :partition) edc)})) (defn ->topics - "Yield a valid object for subscription" + "Yield a valid topic object for subscription given a string, keyword, + regex pattern or collection of strings or keywords." ^Collection [topics] + (assert (or (string? topics) + (keyword? topics) + (instance? Pattern topics) + (and (instance? Collection topics) + (every? (some-fn string? keyword?) topics))) + (str "topics argument must be a string, keyword, regex pattern or " + "collection of strings or keywords, received " topics)) (cond (keyword? topics) [(name topics)] (string? topics) [topics] @@ -428,8 +436,8 @@ :else (throw (ex-info "topics argument is invalid" {:topics topics})))) (defn consumer->driver - "Given a consumer-driver and an optional callback to callback - to call when stopping, yield a consumer driver. + "Given a KafkaConsumer and an optional callback to call when stopping, + yield a consumer driver. The consumer driver implements the following protocols: @@ -466,22 +474,8 @@ (.resume consumer (map ->topic-partition topic-partitions))) (subscribe! [this topics] - (assert (or (string? topics) - (keyword? topics) - (instance? Pattern topics) - (and (instance? Collection topics) - (every? (some-fn string? keyword?) topics))) - (str "topic argument must be a string, keyword, regex pattern or " - "collection of strings or keywords.")) (.subscribe consumer (->topics topics))) (subscribe! [this topics listener] - (assert (or (string? topics) - (keyword? topics) - (instance? Pattern topics) - (and (instance? Collection topics) - (every? (some-fn string? keyword?) topics))) - (str "topic argument must be a string, keyword, regex pattern or " - "collection of strings or keywords.")) (.subscribe consumer (->topics topics) (rebalance-listener listener))) (unsubscribe! [this] (.unsubscribe consumer)) @@ -494,17 +488,17 @@ (map (juxt ->topic-partition ->offset-metadata)) (reduce merge {}))] (.commitSync consumer ^Map offsets))) - (seek! [this topic-partition offset] - (.seek consumer - (->topic-partition topic-partition) - (long offset))) - (position! [this topic-partition] - (.position consumer (->topic-partition topic-partition))) + (seek! [this topic-partition offset] + (.seek consumer + (->topic-partition topic-partition) + (long offset))) + (position! [this topic-partition] + (.position consumer (->topic-partition topic-partition))) (subscription [this] (.subscription consumer)) - GenericDriver - (close! [this] - (.close consumer)) + GenericDriver + (close! [this] + (.close consumer)) MetadataDriver (partitions-for [this topic] (mapv partition-info->data (.partitionsFor consumer topic))) @@ -537,8 +531,9 @@ (map ->header headers)) (defn ->record - "Build a producer record from a clojure map. Leave ProducerRecord instances - untouched." + "Build a ProducerRecord from a clojure map. + Leave ProducerRecord instances untouched." + ^ProducerRecord [payload] (if (instance? ProducerRecord payload) payload @@ -580,7 +575,6 @@ (send! [this topic k v] (.send producer (->record {:key k :value v :topic topic}))) (send! [this topic k v headers] - "Defaults partition and timestamp to 0, if you need to set either, use the single arity version" (.send producer (->record {:key k :value v :topic topic :headers headers}))) (flush! [this] (.flush producer)) From 9c4fee9e1c31984a52d7286541a03b4e687c1d06 Mon Sep 17 00:00:00 2001 From: Francois Date: Wed, 24 Oct 2018 16:22:34 +0100 Subject: [PATCH 2/7] producer: added send-cb! function that takes a callback fn --- README.md | 1 + src/kinsky/client.clj | 50 +++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 49 insertions(+), 2 deletions(-) mode change 100644 => 100755 src/kinsky/client.clj diff --git a/README.md b/README.md index 22faca2..ee66062 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,7 @@ Thanks a lot to these awesome contributors ### 0.1.25 - Removal of the asynchronous façade, transducers should suffice +- Producer: added send-cb! function that takes a callback fn - Minor typo and code refactoring ### 0.1.24 diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj old mode 100644 new mode 100755 index 34a06a2..1ec4f45 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -12,8 +12,10 @@ ConsumerRecords KafkaConsumer OffsetAndMetadata) - (org.apache.kafka.clients.producer KafkaProducer - ProducerRecord) + (org.apache.kafka.clients.producer Callback + KafkaProducer + ProducerRecord + RecordMetadata) (org.apache.kafka.common Node PartitionInfo TopicPartition) @@ -129,6 +131,12 @@ with the following possible keys is expected: `:key`, `:topic`, `:partition`, `:headers`, `:timestamp` and `:value`. ") + (send-cb! [this record cb] [this topic k v cb] [this topic k v headers cb] + "Produce a record on a topic with a callback function of 2 arguments: + - a producer record map as per rm->data, + - an exception thrown during the processing of a record. + One of the two arguments will be nil depending on the send result. + ") (flush! [this] "Ensure that produced messages are flushed.") (init-transactions! [this]) @@ -551,6 +559,34 @@ key value (->headers headers))))) +(defn rm->data + "Yield a clojure representation of a producer RecordMetadata. + The map returned is in the form: + {:topic \"topic-name\" + :partition 0 ;; nil if unknown + :offset 1234567890 + :timestamp 9876543210}" + [^RecordMetadata rm] + {:topic (.topic rm) + :partition (let [partition (.partition rm)] + (when (not= partition RecordMetadata/UNKNOWN_PARTITION) + partition)) + :offset (.offset rm) + :timestamp (.timestamp rm)}) + +(defn ->callback + "Return a producer Callback instance given a function taking 2 arguments, + - a producer record map as per rm->data, + - an exception thrown during the processing of a record. + One of the two argument will be nil depending on the send result." + ^Callback + [f] + (when f + (reify + Callback + (onCompletion [_ record-metadata exception] + (f (some-> record-metadata rm->data) exception))))) + (defn producer->driver "Yield a driver from a Kafka Producer. The producer driver implements the following protocols: @@ -576,6 +612,16 @@ (.send producer (->record {:key k :value v :topic topic}))) (send! [this topic k v headers] (.send producer (->record {:key k :value v :topic topic :headers headers}))) + (send-cb! [this record cb] + (.send producer (->record record) (->callback cb))) + (send-cb! [this topic k v cb] + (.send producer + (->record {:key k :value v :topic topic}) + (->callback cb))) + (send-cb! [this topic k v headers cb] + (.send producer + (->record {:key k :value v :topic topic :headers headers}) + (->callback cb))) (flush! [this] (.flush producer)) (init-transactions! [this] From cc8ab1b8bfa50c433bd90453b0a0823edb2c2b89 Mon Sep 17 00:00:00 2001 From: Francois Date: Tue, 16 Oct 2018 07:50:49 +0100 Subject: [PATCH 3/7] consumer: added seek-beginning! and seek-end! --- README.md | 1 + src/kinsky/client.clj | 22 ++++++++++++++++------ 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index ee66062..50269cc 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ Thanks a lot to these awesome contributors - Removal of the asynchronous façade, transducers should suffice - Producer: added send-cb! function that takes a callback fn +- Consumer: added seek-beginning! and seek-end! - Minor typo and code refactoring ### 0.1.24 diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj index 1ec4f45..d68a8a2 100755 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -117,10 +117,18 @@ (wake-up! [this] "Safely wake-up a consumer which may be blocking during polling.") (seek! [this] [this topic-partition offset] - "Overrides the fetch offsets that the consumer will use on the next poll") - (position! [this] [this topic-partition] + "Overrides the fetch offsets the consumer will use on the next poll. + topic-partition must be a map with `:topic` and `:partition` entries. + and offset must be a long value.") + (seek-beginning! [this] [this topic-partitions] + "Overrides the fetch offsets the consumer will use on the next poll. + topic-partitions must be a seq of maps with `:topic` and `:partition`.") + (seek-end! [this] [this topic-partitions] + "Overrides the fetch offsets the consumer will use on the next poll. + topic-partitions must be a seq of maps with `:topic` and `:partition`.") + (position! [this] [this topic-partition] "Get the offset of the next record that will be fetched (if a record with that offset exists).") - (subscription [this] + (subscription [this] "Currently assigned topics")) (defprotocol ProducerDriver @@ -497,9 +505,11 @@ (reduce merge {}))] (.commitSync consumer ^Map offsets))) (seek! [this topic-partition offset] - (.seek consumer - (->topic-partition topic-partition) - (long offset))) + (.seek consumer (->topic-partition topic-partition) (long offset))) + (seek-beginning! [this topic-partitions] + (.seekToBeginning consumer (map ->topic-partition topic-partitions))) + (seek-end! [this topic-partitions] + (.seekToEnd consumer (map ->topic-partition topic-partitions))) (position! [this topic-partition] (.position consumer (->topic-partition topic-partition))) (subscription [this] From e14be41ac4f8d27baf2a3154fed7c43b8aa2f846 Mon Sep 17 00:00:00 2001 From: Francois Date: Mon, 16 Dec 2019 00:00:00 +0100 Subject: [PATCH 4/7] producer: added abort-transaction! --- README.md | 1 + src/kinsky/client.clj | 13 ++++++++++--- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 50269cc..ad9a783 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ Thanks a lot to these awesome contributors - Removal of the asynchronous façade, transducers should suffice - Producer: added send-cb! function that takes a callback fn +- Producer: added abort-transaction! function - Consumer: added seek-beginning! and seek-end! - Minor typo and code refactoring diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj index d68a8a2..88c5588 100755 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -147,9 +147,14 @@ ") (flush! [this] "Ensure that produced messages are flushed.") - (init-transactions! [this]) - (begin-transaction! [this]) - (commit-transaction! [this])) + (init-transactions! [this] + "To be called once when a `transactional.id` is configured.") + (begin-transaction! [this] + "Initiate a new transaction.") + (commit-transaction! [this] + "Commits the ongoing transaction.") + (abort-transaction! [this] + "Aborts the ongoing transaction.")) (defprotocol GenericDriver (close! [this] [this timeout] @@ -640,6 +645,8 @@ (.beginTransaction producer)) (commit-transaction! [this] (.commitTransaction producer)) + (abort-transaction! [this] + (.abortTransaction producer)) MetadataDriver (partitions-for [this topic] (mapv partition-info->data (.partitionsFor producer topic))) From b106ae9982aa641f9c60c6c0461a90dc6b22b2cd Mon Sep 17 00:00:00 2001 From: Francois Date: Mon, 23 Dec 2019 23:24:36 +0100 Subject: [PATCH 5/7] consumer+producer: added name and counter to support toString --- README.md | 1 + src/kinsky/client.clj | 224 ++++++++++++++++++++++++------------------ 2 files changed, 127 insertions(+), 98 deletions(-) diff --git a/README.md b/README.md index ad9a783..9f0a8a8 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,7 @@ Thanks a lot to these awesome contributors ### 0.1.25 - Removal of the asynchronous façade, transducers should suffice +- Consumer+Producer: added name and counter to support toString - Producer: added send-cb! function that takes a callback fn - Producer: added abort-transaction! function - Consumer: added seek-beginning! and seek-end! diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj index 88c5588..fbc8380 100755 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -456,6 +456,14 @@ (instance? Pattern topics) topics :else (throw (ex-info "topics argument is invalid" {:topics topics})))) +(def consumer-counter (atom 0)) +(defn opts->consumer-id + ^String + [opts] + (str "kinsky-consumer-" (swap! consumer-counter inc) "-" + (get opts "group.id" "") "-" + (get opts "client.id" "") "-")) + (defn consumer->driver "Given a KafkaConsumer and an optional callback to call when stopping, yield a consumer driver. @@ -467,6 +475,7 @@ - `clojure.lang.IDeref`: `deref` to access underlying [KafkaConsumer](http://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/KafkaConsumer.html) instance. + - `java.lang.Object`: `.toString` produces the kinsky consumer `id`. The consumer-driver can also take options: @@ -477,57 +486,61 @@ (consumer->driver consumer nil)) ([^KafkaConsumer consumer {::keys [run-fn consumer-decoder-fn] - :or {consumer-decoder-fn consumer-records->data}}] - (reify - ConsumerDriver - (poll! [this timeout] - (consumer-decoder-fn (.poll consumer (java.time.Duration/ofMillis timeout)))) - (stop! [this] - (stop! this 0)) - (stop! [this timeout] - (when (fn? run-fn) - (run-fn)) - (.wakeup consumer)) - (pause! [this topic-partitions] - (.pause consumer - (map ->topic-partition topic-partitions))) - (resume! [this topic-partitions] - (.resume consumer - (map ->topic-partition topic-partitions))) - (subscribe! [this topics] - (.subscribe consumer (->topics topics))) - (subscribe! [this topics listener] - (.subscribe consumer (->topics topics) (rebalance-listener listener))) - (unsubscribe! [this] - (.unsubscribe consumer)) - (wake-up! [this] - (.wakeup consumer)) - (commit! [this] - (.commitSync consumer)) - (commit! [this topic-offsets] - (let [offsets (->> topic-offsets - (map (juxt ->topic-partition ->offset-metadata)) - (reduce merge {}))] - (.commitSync consumer ^Map offsets))) - (seek! [this topic-partition offset] - (.seek consumer (->topic-partition topic-partition) (long offset))) - (seek-beginning! [this topic-partitions] - (.seekToBeginning consumer (map ->topic-partition topic-partitions))) - (seek-end! [this topic-partitions] - (.seekToEnd consumer (map ->topic-partition topic-partitions))) - (position! [this topic-partition] - (.position consumer (->topic-partition topic-partition))) - (subscription [this] - (.subscription consumer)) - GenericDriver - (close! [this] - (.close consumer)) - MetadataDriver - (partitions-for [this topic] - (mapv partition-info->data (.partitionsFor consumer topic))) - clojure.lang.IDeref - (deref [this] - consumer)))) + :or {consumer-decoder-fn consumer-records->data} + :as config}] + (let [id (opts->consumer-id config)] + (reify + ConsumerDriver + (poll! [this timeout] + (consumer-decoder-fn (.poll consumer (java.time.Duration/ofMillis timeout)))) + (stop! [this] + (stop! this 0)) + (stop! [this timeout] + (when (fn? run-fn) + (run-fn)) + (.wakeup consumer)) + (pause! [this topic-partitions] + (.pause consumer + (map ->topic-partition topic-partitions))) + (resume! [this topic-partitions] + (.resume consumer + (map ->topic-partition topic-partitions))) + (subscribe! [this topics] + (.subscribe consumer (->topics topics))) + (subscribe! [this topics listener] + (.subscribe consumer (->topics topics) (rebalance-listener listener))) + (unsubscribe! [this] + (.unsubscribe consumer)) + (wake-up! [this] + (.wakeup consumer)) + (commit! [this] + (.commitSync consumer)) + (commit! [this topic-offsets] + (let [offsets (->> topic-offsets + (map (juxt ->topic-partition ->offset-metadata)) + (reduce merge {}))] + (.commitSync consumer ^Map offsets))) + (seek! [this topic-partition offset] + (.seek consumer (->topic-partition topic-partition) (long offset))) + (seek-beginning! [this topic-partitions] + (.seekToBeginning consumer (map ->topic-partition topic-partitions))) + (seek-end! [this topic-partitions] + (.seekToEnd consumer (map ->topic-partition topic-partitions))) + (position! [this topic-partition] + (.position consumer (->topic-partition topic-partition))) + (subscription [this] + (.subscription consumer)) + GenericDriver + (close! [this] + (.close consumer)) + MetadataDriver + (partitions-for [this topic] + (mapv partition-info->data (.partitionsFor consumer topic))) + clojure.lang.IDeref + (deref [this] + consumer) + Object + (toString [this] id))))) (defn safe-poll! "Implementation of poll which disregards wake-up exceptions" @@ -602,6 +615,14 @@ (onCompletion [_ record-metadata exception] (f (some-> record-metadata rm->data) exception))))) +(def producer-counter (atom 0)) +(defn opts->producer-id + ^String + [opts] + (str "kinsky-producer-" (swap! producer-counter inc) "-" + (get opts "transaction.id" "") "-" + (get opts "client.id" "") "-")) + (defn producer->driver "Yield a driver from a Kafka Producer. The producer driver implements the following protocols: @@ -610,49 +631,55 @@ - [MetadataDriver](#var-MetadataDriver) - `clojure.lang.IDeref`: `deref` to access underlying [KafkaProducer](http://kafka.apache.org/090/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html) - instance." - [^KafkaProducer producer] - (reify - GenericDriver - (close! [this] - (.close producer)) - (close! [this timeout] - (if (nil? timeout) - (.close producer) - (.close producer (long timeout) TimeUnit/MILLISECONDS))) - ProducerDriver - (send! [this record] - (.send producer (->record record))) - (send! [this topic k v] - (.send producer (->record {:key k :value v :topic topic}))) - (send! [this topic k v headers] - (.send producer (->record {:key k :value v :topic topic :headers headers}))) - (send-cb! [this record cb] - (.send producer (->record record) (->callback cb))) - (send-cb! [this topic k v cb] - (.send producer - (->record {:key k :value v :topic topic}) - (->callback cb))) - (send-cb! [this topic k v headers cb] - (.send producer - (->record {:key k :value v :topic topic :headers headers}) - (->callback cb))) - (flush! [this] - (.flush producer)) - (init-transactions! [this] - (.initTransactions producer)) - (begin-transaction! [this] - (.beginTransaction producer)) - (commit-transaction! [this] - (.commitTransaction producer)) - (abort-transaction! [this] - (.abortTransaction producer)) - MetadataDriver - (partitions-for [this topic] - (mapv partition-info->data (.partitionsFor producer topic))) - clojure.lang.IDeref - (deref [this] - producer))) + instance. + - `java.lang.Object`: `.toString` produces the kinsky producer `id`." + ([^KafkaProducer producer] + (producer->driver producer nil)) + ([^KafkaProducer producer config] + (let [id (opts->producer-id config)] + (reify + GenericDriver + (close! [this] + (.close producer)) + (close! [this timeout] + (if (nil? timeout) + (.close producer) + (.close producer (long timeout) TimeUnit/MILLISECONDS))) + ProducerDriver + (send! [this record] + (.send producer (->record record))) + (send! [this topic k v] + (.send producer (->record {:key k :value v :topic topic}))) + (send! [this topic k v headers] + (.send producer (->record {:key k :value v :topic topic :headers headers}))) + (send-cb! [this record cb] + (.send producer (->record record) (->callback cb))) + (send-cb! [this topic k v cb] + (.send producer + (->record {:key k :value v :topic topic}) + (->callback cb))) + (send-cb! [this topic k v headers cb] + (.send producer + (->record {:key k :value v :topic topic :headers headers}) + (->callback cb))) + (flush! [this] + (.flush producer)) + (init-transactions! [this] + (.initTransactions producer)) + (begin-transaction! [this] + (.beginTransaction producer)) + (commit-transaction! [this] + (.commitTransaction producer)) + (abort-transaction! [this] + (.abortTransaction producer)) + MetadataDriver + (partitions-for [this topic] + (mapv partition-info->data (.partitionsFor producer topic))) + clojure.lang.IDeref + (deref [this] + producer) + Object + (toString [this] id))))) (defn producer "Create a producer from a configuration and optional serializers. @@ -660,16 +687,17 @@ and values. If none are provided, the configuration is expected to hold serializer class names." ([config] - (producer->driver (KafkaProducer. (opts->props config)))) + (producer->driver (KafkaProducer. (opts->props config)) config)) ([config serializer] (producer->driver (KafkaProducer. (opts->props config) (->serializer serializer) - (->serializer serializer)))) + (->serializer serializer)) + config)) ([config kserializer vserializer] (producer->driver (KafkaProducer. (opts->props config) (->serializer kserializer) - (->serializer vserializer))))) - + (->serializer vserializer)) + config))) (defn consumer "Create a consumer from a configuration and optional deserializers. If a callback is given, call it when stopping the consumer. From 7629c0af903ac9b994d4e872f60d90d91b23538e Mon Sep 17 00:00:00 2001 From: Francois Date: Mon, 23 Dec 2019 23:35:03 +0100 Subject: [PATCH 6/7] consumer: added close! with timeout argument --- src/kinsky/client.clj | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/kinsky/client.clj b/src/kinsky/client.clj index fbc8380..74583dc 100755 --- a/src/kinsky/client.clj +++ b/src/kinsky/client.clj @@ -533,6 +533,10 @@ GenericDriver (close! [this] (.close consumer)) + (close! [this timeout] + (if (nil? timeout) + (.close consumer) + (.close consumer (long timeout) TimeUnit/MILLISECONDS))) MetadataDriver (partitions-for [this topic] (mapv partition-info->data (.partitionsFor consumer topic))) From d0a5de3d1d8ab76e85a405fec3b04b9c55d12fab Mon Sep 17 00:00:00 2001 From: Francois Date: Mon, 23 Dec 2019 23:18:14 +0100 Subject: [PATCH 7/7] minor test improvements --- test/kinsky/client_test.clj | 20 +++++++++++--------- 1 file changed, 11 insertions(+), 9 deletions(-) diff --git a/test/kinsky/client_test.clj b/test/kinsky/client_test.clj index a0548f6..d716951 100644 --- a/test/kinsky/client_test.clj +++ b/test/kinsky/client_test.clj @@ -1,5 +1,5 @@ (ns kinsky.client-test - (:require [clojure.test :refer :all :as t] + (:require [clojure.test :refer :all :as t] [clojure.pprint :as pp] [kinsky.client :as client] [kinsky.embedded :as e])) @@ -32,22 +32,22 @@ (testing "string serializer" (is (= "foo" (String. - (.serialize (client/string-serializer) "" "foo"))))) + (.serialize (client/string-serializer) "" "foo"))))) (testing "keyword serializer" (is (= "foo" (String. - (.serialize (client/keyword-serializer) "" :foo))))) + (.serialize (client/keyword-serializer) "" :foo))))) (testing "edn serializer" (is (= "{:a :b, :c :d}" (String. - (.serialize (client/edn-serializer) "" {:a :b :c :d}))))) + (.serialize (client/edn-serializer) "" {:a :b :c :d}))))) (testing "json serializer" (is (= "[0,1,2]" (String. - (.serialize (client/json-serializer) "" [0 1 2])))))) + (.serialize (client/json-serializer) "" [0 1 2])))))) (deftest deserializer (testing "string deserializer" @@ -69,12 +69,14 @@ (.getBytes "{\"a\": \"b\", \"c\": \"d\"}")))))) (deftest config-props - (testing "valid configuration properties" - (is (= {"foo.bar" "0"} - (client/opts->props {:foo.bar 0}))))) + (testing "configuration properties" + (is (= {"foo.bar" "0" "foo.baz" "1"} + (client/opts->props {:foo.bar 0 + "foo.baz" "1" + :qualified/kw :discarded}))))) (deftest rebalance-listener - (testing "idempotency" + (testing "identity when given a ConsumerRebalanceListener" (let [sink (client/rebalance-listener (fn [& _]))] (is (= sink (client/rebalance-listener sink)))))