From 2210ee587d73dff06ac0711280b0ea8aae3ffa10 Mon Sep 17 00:00:00 2001 From: Shwetank Dave Date: Wed, 16 Jan 2019 12:27:47 -0500 Subject: Removing conform from kafkaRecords Change-Id: Id3648a7286556e683b50da9f3db4bcec62005794 Issue-ID: AAI-797 Signed-off-by: Shwetank Dave --- src/chameleon/kafka.clj | 1 - src/chameleon/specs.clj | 11 ++++++++++- 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]))) -- cgit 1.2.3-korg