summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--src/chameleon/kafka.clj1
-rw-r--r--src/chameleon/specs.clj11
2 files changed, 10 insertions, 2 deletions
diff --git a/src/chameleon/kafka.clj b/src/chameleon/kafka.clj
index bdfc07d..357d68a 100644
--- a/src/chameleon/kafka.clj
+++ b/src/chameleon/kafka.clj
@@ -99,7 +99,6 @@
(loop []
(let [consumer-records (.poll consumer 0)]
(when (and (->> consumer-records
- (s/conform :kafka/consumer-record)
(map consumer-record->clojure)
(send-with-sla-timeout chan sla-timeout))
(try (.commitSync consumer)
diff --git a/src/chameleon/specs.clj b/src/chameleon/specs.clj
index 760b30a..5a8a896 100644
--- a/src/chameleon/specs.clj
+++ b/src/chameleon/specs.clj
@@ -7,7 +7,8 @@
[clojure.string :as str]
[cheshire.core :as json]
[chameleon.kafka :as ck])
- (:import [org.apache.kafka.clients.consumer ConsumerRecord]
+ (:import [org.apache.kafka.clients.consumer ConsumerRecord ConsumerRecords ]
+ [org.apache.kafka.common TopicPartition]
[java.util Properties]))
(s/def ::host string?)
@@ -102,6 +103,11 @@
(s/def :kafka/key ::string)
(s/def :kafka/value ::string)
+(s/def :kafka/topic-partition (s/spec #(instance? TopicPartition %) :gen (fn [] (->> (s/cat :topic :kafka/topic
+ :partition :kafka/partition)
+ s/gen
+ (gen/fmap (fn [[topic partition]]
+ (TopicPartition. topic partition)))))))
(s/def :kafka/consumer-record (s/spec #(instance? ConsumerRecord %)
:gen (fn [] (->> (s/cat :topic :kafka/topic
:partition :kafka/partition
@@ -111,6 +117,9 @@
s/gen
(gen/fmap (fn [[topic partition offset key value]]
(ConsumerRecord. topic partition offset key value)))))))
+(s/def :kafka/consumer-records (s/spec #(instance? ConsumerRecords %)
+ :gen (fn [] (gen/fmap (fn [k] (ConsumerRecords. k))
+ (s/gen (s/map-of :kafka/topic-partition (s/coll-of :kafka/consumer-record)))))))
(s/def :kafka/clojure-consumer-record (s/spec (s/keys :req-un [:kafka/topic :kafka/partition
:kafka/offset :kafka/key :kafka/value])))