From 7f6092fd4bcc5d5e4dc7f58fe9cbcd0c1f0dfe94 Mon Sep 17 00:00:00 2001 From: Greg Haskins Date: Sun, 29 May 2022 13:11:34 -0400 Subject: [PATCH 1/4] Port to kafka v3.2.0 and update deps Signed-off-by: Greg Haskins --- project.clj | 58 +++++++++++----------- src/jackdaw/data/common_record.clj | 2 +- src/jackdaw/data/consumer.clj | 11 ++-- src/jackdaw/data/producer.clj | 24 +-------- src/jackdaw/streams/interop.clj | 10 ++-- src/jackdaw/streams/lambdas.clj | 10 ++-- src/jackdaw/test/transports/kafka.clj | 3 +- src/jackdaw/test/transports/mock.clj | 55 ++++++++++++-------- test/jackdaw/streams_test.clj | 24 ++++----- test/jackdaw/test/transports/mock_test.clj | 18 +++---- 10 files changed, 100 insertions(+), 115 deletions(-) diff --git a/project.clj b/project.clj index cdc9d61f..7730ad10 100644 --- a/project.clj +++ b/project.clj @@ -8,39 +8,39 @@ :repositories [["confluent" {:url "https://packages.confluent.io/maven/"}] ["mulesoft" {:url "https://repository.mulesoft.org/nexus/content/repositories/public/"}]] - :dependencies [[aleph "0.4.6"] + :dependencies [[aleph "0.4.7"] [danlentz/clj-uuid "0.1.9" :exclusions [primitive-math]] ;; Confluent does paired releases with Kafka, this should tie ;; off with the kafka version. ;; See https://docs.confluent.io/current/release-notes.html - [io.confluent/kafka-schema-registry-client "6.1.1" + [io.confluent/kafka-schema-registry-client "7.1.1" :exclusions [com.fasterxml.jackson.core/jackson-databind]] - [io.confluent/kafka-avro-serializer "6.1.1"] - [io.confluent/kafka-json-schema-serializer "6.1.1"] - [org.apache.kafka/kafka-clients "2.8.0"] - [org.apache.kafka/kafka-streams "2.8.0"] - [org.apache.kafka/kafka-streams-test-utils "2.8.0"] + [io.confluent/kafka-avro-serializer "7.1.1"] + [io.confluent/kafka-json-schema-serializer "7.1.1"] + [org.apache.kafka/kafka-clients "3.2.0"] + [org.apache.kafka/kafka-streams "3.2.0"] + [org.apache.kafka/kafka-streams-test-utils "3.2.0"] - [org.clojure/clojure "1.10.1" :scope "provided"] + [org.clojure/clojure "1.11.1" :scope "provided"] [org.clojure/java.data "1.0.95"] - [org.clojure/data.json "0.2.6"] - [org.clojure/data.fressian "0.2.1"] - [org.clojure/tools.logging "0.4.1"] - [org.clojure/core.cache "0.7.2"] - [metosin/jsonista "0.3.3"] + [org.clojure/data.json "2.4.0"] + [org.clojure/data.fressian "1.0.0"] + [org.clojure/tools.logging "1.2.4"] + [org.clojure/core.cache "1.0.225"] + [metosin/jsonista "0.3.5"] ;; Use specific netty version to avoid critical CVE ;; pulled by Aleph v0.4.6 (last stable version) - [io.netty/netty-transport "4.1.68.Final"] - [io.netty/netty-transport-native-epoll "4.1.68.Final"] - [io.netty/netty-codec "4.1.68.Final"] - [io.netty/netty-codec-http "4.1.68.Final"] - [io.netty/netty-handler "4.1.68.Final"] - [io.netty/netty-handler-proxy "4.1.68.Final"] - [io.netty/netty-resolver "4.1.68.Final"] - [io.netty/netty-resolver-dns "4.1.68.Final"] + [io.netty/netty-transport "4.1.77.Final"] + [io.netty/netty-transport-native-epoll "4.1.77.Final"] + [io.netty/netty-codec "4.1.77.Final"] + [io.netty/netty-codec-http "4.1.77.Final"] + [io.netty/netty-handler "4.1.77.Final"] + [io.netty/netty-handler-proxy "4.1.77.Final"] + [io.netty/netty-resolver "4.1.77.Final"] + [io.netty/netty-resolver-dns "4.1.77.Final"] ;; Use specific commons-compress version to avoid ;; CVE-2021-36090 pulled by avro 1.9.2 @@ -87,14 +87,14 @@ :resource-paths ["test/resources"] :injections [(require 'io.aviso.logging.setup)] - :dependencies [[io.aviso/logging "0.3.2"] - [org.apache.kafka/kafka-streams-test-utils "2.8.0"] - [org.apache.kafka/kafka-clients "2.8.0" :classifier "test"] - [org.clojure/test.check "0.9.0"] - [org.apache.kafka/kafka_2.13 "2.8.0"] - [lambdaisland/kaocha "0.0-529"] - [lambdaisland/kaocha-cloverage "0.0-32"] - [lambdaisland/kaocha-junit-xml "0.0-70"]]} + :dependencies [[io.aviso/logging "1.0"] + [org.apache.kafka/kafka-streams-test-utils "3.2.0"] + [org.apache.kafka/kafka-clients "3.2.0" :classifier "test"] + [org.clojure/test.check "1.1.1"] + [org.apache.kafka/kafka_2.13 "3.2.0"] + [lambdaisland/kaocha "1.66.1034"] + [lambdaisland/kaocha-cloverage "1.0.75"] + [lambdaisland/kaocha-junit-xml "0.0.76"]]} ;; This is not in fact what lein defines repl to be :repl diff --git a/src/jackdaw/data/common_record.clj b/src/jackdaw/data/common_record.clj index 4363964d..1e0f3772 100644 --- a/src/jackdaw/data/common_record.clj +++ b/src/jackdaw/data/common_record.clj @@ -24,7 +24,7 @@ (defn ->TimestampType "Given a keyword being a datafied Kafka `TimestampType`, return the equivalent `TimestampType` instance." - [kw] + ^TimestampType [kw] (case kw :timestamp-type/create TimestampType/CREATE_TIME :timestamp-type/log-append TimestampType/LOG_APPEND_TIME diff --git a/src/jackdaw/data/consumer.clj b/src/jackdaw/data/consumer.clj index 04a8c485..389271b2 100644 --- a/src/jackdaw/data/consumer.clj +++ b/src/jackdaw/data/consumer.clj @@ -7,7 +7,8 @@ (import '[org.apache.kafka.clients.consumer ConsumerRecord OffsetAndTimestamp] 'org.apache.kafka.common.header.Headers - 'org.apache.kafka.common.record.TimestampType) + 'org.apache.kafka.common.record.TimestampType + 'java.util.Optional) (set! *warn-on-reflection* true) @@ -17,18 +18,18 @@ Convenient for testing the consumer API and its helpers." ^ConsumerRecord [{:keys [:topic-name]} partition offset ts ts-type key-size value-size key value ^Headers headers] - (ConsumerRecord. topic-name + (ConsumerRecord. ^String topic-name (int partition) (long offset) (long ts) (if (keyword? ts-type) (->TimestampType ts-type) ^TimestampType ts-type) - nil ;; Deprecated checksum (int key-size) (int value-size) key value - headers)) + ^Headers headers + (Optional/empty))) (defn map->ConsumerRecord "Given a `::consumer-record`, build an equivalent `ConsumerRecord`. @@ -56,8 +57,6 @@ :headers (.headers r) :partition (.partition r) :timestamp (.timestamp r) - ;; Deprecated field - ;; :checksum (.checksum r) :timestamp-type (TimestampType->data (.timestampType r)) :offset (.offset r) :serialized-key-size (.serializedKeySize r) diff --git a/src/jackdaw/data/producer.clj b/src/jackdaw/data/producer.clj index 87c61f32..c87ea486 100644 --- a/src/jackdaw/data/producer.clj +++ b/src/jackdaw/data/producer.clj @@ -76,17 +76,11 @@ previous abstraction that there's a \"offset\" field which is absolute, an additional arity is provided which lets the user construct a record with a base offset and a relative offset of 0 so - that the metadata's apparent offset is predictable. - - Note that as the checksum is deprecated, by default it is not - required. The third arity allows a user to provide a checksum. This - arity may be removed in the future pending further breaking changes - to the Kafka APIs." + that the metadata's apparent offset is predictable." ([t partition offset timestamp key-size value-size] (RecordMetadata. (->TopicPartition t partition) offset 0 ;; Force absolute offset timestamp - nil ;; No checksum, it's deprecated ^Integer (when key-size (int key-size)) ^Integer (when value-size (int value-size)))) ([t partition base-offset relative-offset timestamp @@ -95,16 +89,6 @@ base-offset relative-offset ;; Full offset control timestamp - nil ;; No checksum, it's depreciated - ^Integer (when key-size (int key-size)) - ^Integer (when value-size (int value-size)))) - ([t partition base-offset relative-offset timestamp checksum - key-size value-size] - (RecordMetadata. (->TopicPartition t partition) - base-offset - relative-offset ;; Full offset control - timestamp - checksum ;; Have fun I guess ^Integer (when key-size (int key-size)) ^Integer (when value-size (int value-size))))) @@ -124,12 +108,6 @@ {:topic-name (.topic rm) :partition (.partition rm) :timestamp (.timestamp rm) - - ;; As of Kafka 0.11.0 the checksum is deprecated. It is no longer - ;; part of Kafka wire protocol, while the brokers may use - ;; checksuming at reset to ensure message integrity. - - ;; :jackdaw.sent-record/checksum (.checksum rm) :offset (.offset rm) :serialized-key-size (.serializedKeySize rm) :serialized-value-size (.serializedValueSize rm)}) diff --git a/src/jackdaw/streams/interop.clj b/src/jackdaw/streams/interop.clj index e853087b..16baee87 100644 --- a/src/jackdaw/streams/interop.clj +++ b/src/jackdaw/streams/interop.clj @@ -14,7 +14,7 @@ [org.apache.kafka.streams StreamsBuilder] [org.apache.kafka.streams.kstream - Aggregator Consumed GlobalKTable Grouped Initializer Joined + Aggregator Consumed GlobalKTable Grouped Initializer Joined StreamJoined JoinWindows KGroupedStream KGroupedTable KStream KTable KeyValueMapper Materialized Merger Predicate Printed Produced Reducer SessionWindowedKStream SessionWindows @@ -266,7 +266,7 @@ ^KStream (kstream* other-kstream) ^ValueJoiner (value-joiner value-joiner-fn) ^JoinWindows windows - (Joined/with key-serde this-value-serde other-value-serde)))) + (StreamJoined/with key-serde this-value-serde other-value-serde)))) (left-join-windowed [_ other-kstream value-joiner-fn windows] @@ -285,7 +285,7 @@ ^KStream (kstream* other-kstream) ^ValueJoiner (value-joiner value-joiner-fn) ^JoinWindows windows - (Joined/with key-serde value-serde other-value-serde)))) + (StreamJoined/with key-serde value-serde other-value-serde)))) (map [_ key-value-mapper-fn] @@ -315,13 +315,13 @@ ^KStream (kstream* other-kstream) ^ValueJoiner (value-joiner value-joiner-fn) ^JoinWindows windows - (Joined/with key-serde value-serde other-value-serde)))) + (StreamJoined/with key-serde value-serde other-value-serde)))) (process! [_ processor-supplier-fn state-store-names] (.process ^KStream kstream ^ProcessorSupplier (processor-supplier processor-supplier-fn) - (into-array String state-store-names))) + ^"[Ljava.lang.String;" (into-array String state-store-names))) (select-key [_ select-key-value-mapper-fn] diff --git a/src/jackdaw/streams/lambdas.clj b/src/jackdaw/streams/lambdas.clj index a6f4d0b6..31beb6d1 100644 --- a/src/jackdaw/streams/lambdas.clj +++ b/src/jackdaw/streams/lambdas.clj @@ -7,7 +7,9 @@ Merger Predicate Reducer Transformer TransformerSupplier ValueJoiner ValueMapper ValueTransformer ValueTransformerSupplier] [org.apache.kafka.streams.processor - Processor ProcessorSupplier StreamPartitioner])) + StreamPartitioner] + [org.apache.kafka.streams.processor.api + Processor ProcessorSupplier])) (set! *warn-on-reflection* true) @@ -147,8 +149,8 @@ (close [_]) (init [_ processor-context] (reset! context processor-context)) - (process [_ key message] - (processor-fn @context key message))) + (process [_ record] + (processor-fn @context (.key record) (.value record)))) (defn processor "Packages up a Clojure fn as a kstream processor." @@ -162,7 +164,7 @@ (defn processor-supplier "Packages up a Clojure fn in a kstream processor supplier." - [processor-fn] + ^ProcessorSupplier [processor-fn] (FnProcessorSupplier. processor-fn)) (deftype FnTransformerSupplier [transformer-supplier-fn] diff --git a/src/jackdaw/test/transports/kafka.clj b/src/jackdaw/test/transports/kafka.clj index 55f9598a..c33bdbfd 100644 --- a/src/jackdaw/test/transports/kafka.clj +++ b/src/jackdaw/test/transports/kafka.clj @@ -70,8 +70,7 @@ "Clojurize the ConsumerRecord returned from consuming a kafka record" [^ConsumerRecord consumer-record] (when consumer-record - {:checksum (.checksum consumer-record) - :key (.key consumer-record) + {:key (.key consumer-record) :offset (.offset consumer-record) :partition (.partition consumer-record) :serializedKeySize (.serializedKeySize consumer-record) diff --git a/src/jackdaw/test/transports/mock.clj b/src/jackdaw/test/transports/mock.clj index a2440fa5..850221ab 100644 --- a/src/jackdaw/test/transports/mock.clj +++ b/src/jackdaw/test/transports/mock.clj @@ -2,15 +2,20 @@ (:require [clojure.stacktrace :as stacktrace] [clojure.tools.logging :as log] + [jackdaw.serdes.fn :as jfn] + [jackdaw.streams.mock :as smock] [jackdaw.test.journal :as j] [jackdaw.test.transports :as t :refer [deftransport]] - [jackdaw.test.serde :refer [byte-array-deserializer + [jackdaw.test.serde :refer [byte-array-serde apply-serializers apply-deserializers serde-map]] [manifold.stream :as s] [manifold.deferred :as d]) (:import - (org.apache.kafka.common.record TimestampType) - (org.apache.kafka.clients.consumer ConsumerRecord))) + (org.apache.kafka.common.record TimestampType) + (org.apache.kafka.clients.consumer ConsumerRecord) + (org.apache.kafka.clients.producer ProducerRecord) + (org.apache.kafka.common.header.internals RecordHeaders) + (java.util Optional))) (set! *warn-on-reflection* false) @@ -40,20 +45,23 @@ node by the TopologyTestDriver" [_topic-config] (fn [m] - (let [record (ConsumerRecord. (get-in m [:topic :topic-name]) + (let [record (ConsumerRecord. ^String (get-in m [:topic :topic-name]) (int -1) (long -1) - (:timestamp m) + (long (:timestamp m)) TimestampType/CREATE_TIME, - (long ConsumerRecord/NULL_CHECKSUM) - (if-let [k (:key m)] - (count k) - 0) - (if-let [v (:value m)] - (count v) - 0) + (int + (if-let [k (:key m)] + (count k) + 0)) + (int + (if-let [v (:value m)] + (count v) + 0)) (:key m) - (:value m))] + (:value m) + (RecordHeaders.) + (Optional/empty))] (set-headers record (:headers m)) (assoc m :input-record record)))) @@ -79,13 +87,12 @@ [messages topic-config] (fn [driver] (let [fetch (fn [[k t]] - {:topic k - :output (loop [collected []] - (if-let [o (.readOutput driver (:topic-name t) - byte-array-deserializer - byte-array-deserializer)] - (recur (conj collected o)) - collected))}) + (let [topic-name (:topic-name t)] + {:topic k + :output (loop [collected []] + (if-let [{:keys [key value]} (smock/consume driver (assoc byte-array-serde :topic-name topic-name))] + (recur (conj collected (ProducerRecord. topic-name key value))) + collected))})) topic-batches (->> topic-config (map fetch) (remove #(empty? (:output %))) @@ -160,13 +167,19 @@ {:messages messages :process process})) +(def identity-serializer (jfn/new-serializer {:serialize (fn [_ _ data] data)})) + (deftransport :mock [{:keys [driver topics]}] (let [serdes (serde-map topics) test-consumer (mock-consumer driver topics (get serdes :deserializers)) record-fn (fn [input-record] (try - (.pipeInput driver input-record) + (let [input-topic (.createInputTopic driver + (.topic input-record) + identity-serializer ;; already serialized in mock-producer + identity-serializer)] + (.pipeInput input-topic (.key input-record) (.value input-record))) (catch Exception e (let [trace (with-out-str (stacktrace/print-cause-trace e))] diff --git a/test/jackdaw/streams_test.clj b/test/jackdaw/streams_test.clj index 94a95e3e..c40f4be1 100644 --- a/test/jackdaw/streams_test.clj +++ b/test/jackdaw/streams_test.clj @@ -313,7 +313,7 @@ (let [topic-a (mock/topic "topic-a") topic-b (mock/topic "topic-b") topic-c (mock/topic "topic-c") - windows (JoinWindows/of 1000) + windows (JoinWindows/ofTimeDifferenceWithNoGrace (Duration/ofMillis 1000)) driver (mock/build-driver (fn [builder] (let [left-kstream (k/kstream builder topic-a) right-kstream (k/kstream builder topic-b)] @@ -368,7 +368,7 @@ (let [topic-a (mock/topic "topic-a") topic-b (mock/topic "topic-b") topic-c (mock/topic "topic-c") - windows (JoinWindows/of 1000) + windows (JoinWindows/of (Duration/ofMillis 1000)) driver (mock/build-driver (fn [builder] (let [left-kstream (k/kstream builder topic-a) right-kstream (k/kstream builder topic-b)] @@ -727,9 +727,7 @@ ;; default is used: 24h - window window (Duration/ofMillis 100) grace (Duration/ofMillis 1) - time-windows (-> window - TimeWindows/of - (.grace grace))] + time-windows (TimeWindows/ofSizeAndGrace window grace)] (with-open [driver (mock/build-driver (fn [builder] (-> builder @@ -763,9 +761,7 @@ window (Duration/ofMillis 100) grace (Duration/ofMillis 1) - time-windows (-> window - TimeWindows/of - (.grace grace)) + time-windows (TimeWindows/ofSizeAndGrace window grace) max-records 2] (with-open [driver (mock/build-driver (fn [builder] @@ -801,9 +797,7 @@ window (Duration/ofMillis 100) grace (Duration/ofMillis 1) - time-windows (-> window - TimeWindows/of - (.grace grace)) + time-windows (TimeWindows/ofSizeAndGrace window grace) max-records 2] (with-open [driver (mock/build-driver (fn [builder] @@ -1060,7 +1054,7 @@ (-> builder (k/kstream topic-a) (k/group-by (fn [[k _v]] (long (/ k 10))) topic-a) - (k/window-by-time (TimeWindows/of 1000)) + (k/window-by-time (TimeWindows/ofSizeWithNoGrace (Duration/ofMillis 1000))) (k/reduce + topic-a) (k/to-kstream) (k/map (fn [[k v]] [(.key k) v])) @@ -1084,7 +1078,7 @@ (-> builder (k/kstream topic-a) (k/group-by-key) - (k/window-by-time (TimeWindows/of 1000)) + (k/window-by-time (TimeWindows/ofSizeWithNoGrace (Duration/ofMillis 1000))) (k/reduce + topic-a) (k/to-kstream) (k/map (fn [[k v]] [(.key k) v])) @@ -1108,7 +1102,7 @@ (-> builder (k/kstream topic-a) (k/group-by (fn [[k _v]] (long (/ k 10))) topic-a) - (k/window-by-session (SessionWindows/with 1000)) + (k/window-by-session (SessionWindows/ofInactivityGapWithNoGrace (Duration/ofMillis 1000))) (k/reduce + topic-a) (k/to-kstream) (k/map (fn [[k v]] [(.key k) v])) @@ -1135,7 +1129,7 @@ (-> builder (k/kstream topic-a) (k/group-by-key) - (k/window-by-session (SessionWindows/with 1000)) + (k/window-by-session (SessionWindows/ofInactivityGapWithNoGrace (Duration/ofMillis 1000))) (k/aggregate (constantly 0) (fn [agg [_k v]] (+ agg v)) diff --git a/test/jackdaw/test/transports/mock_test.clj b/test/jackdaw/test/transports/mock_test.clj index 05115a4a..92022127 100644 --- a/test/jackdaw/test/transports/mock_test.clj +++ b/test/jackdaw/test/transports/mock_test.clj @@ -9,8 +9,8 @@ [jackdaw.test.serde :as serde] [manifold.stream :as s]) (:import - (java.util Properties) - (org.apache.kafka.streams TopologyTestDriver))) + (java.util Properties) + (org.apache.kafka.streams TopologyTestDriver Topology))) (set! *warn-on-reflection* false) @@ -45,14 +45,14 @@ (defn test-driver [f app-config] (let [builder (k/streams-builder) - topology (let [builder (f builder)] + ^Topology topology (let [builder (f builder)] (-> (k/streams-builder* builder) - (.build)))] - (TopologyTestDriver. topology - (let [props (Properties.)] - (doseq [[k v] app-config] - (.setProperty props k v)) - props)))) + (.build))) + ^Properties props (let [props (Properties.)] + (doseq [[k v] app-config] + (.setProperty props k v)) + props)] + (TopologyTestDriver. topology props))) (defn mock-transport [] From 0a6497eada8026ba7c53797c3018d44ccd30793a Mon Sep 17 00:00:00 2001 From: Chris McDevitt Date: Fri, 3 Mar 2023 17:41:06 +0000 Subject: [PATCH 2/4] deps: bump some other kafka deps, aleph, jsonista latest aleph, no need to pin netty anymore Aleph 0.6.1 uses Netty 4.1.87.Final --- project.clj | 31 ++++++++----------------------- 1 file changed, 8 insertions(+), 23 deletions(-) diff --git a/project.clj b/project.clj index 7730ad10..99334f12 100644 --- a/project.clj +++ b/project.clj @@ -8,17 +8,17 @@ :repositories [["confluent" {:url "https://packages.confluent.io/maven/"}] ["mulesoft" {:url "https://repository.mulesoft.org/nexus/content/repositories/public/"}]] - :dependencies [[aleph "0.4.7"] + :dependencies [[aleph "0.6.1"] [danlentz/clj-uuid "0.1.9" :exclusions [primitive-math]] ;; Confluent does paired releases with Kafka, this should tie ;; off with the kafka version. ;; See https://docs.confluent.io/current/release-notes.html - [io.confluent/kafka-schema-registry-client "7.1.1" + [io.confluent/kafka-schema-registry-client "7.3.2" :exclusions [com.fasterxml.jackson.core/jackson-databind]] - [io.confluent/kafka-avro-serializer "7.1.1"] - [io.confluent/kafka-json-schema-serializer "7.1.1"] + [io.confluent/kafka-avro-serializer "7.3.2"] + [io.confluent/kafka-json-schema-serializer "7.3.2"] [org.apache.kafka/kafka-clients "3.2.0"] [org.apache.kafka/kafka-streams "3.2.0"] [org.apache.kafka/kafka-streams-test-utils "3.2.0"] @@ -29,22 +29,7 @@ [org.clojure/data.fressian "1.0.0"] [org.clojure/tools.logging "1.2.4"] [org.clojure/core.cache "1.0.225"] - [metosin/jsonista "0.3.5"] - - ;; Use specific netty version to avoid critical CVE - ;; pulled by Aleph v0.4.6 (last stable version) - [io.netty/netty-transport "4.1.77.Final"] - [io.netty/netty-transport-native-epoll "4.1.77.Final"] - [io.netty/netty-codec "4.1.77.Final"] - [io.netty/netty-codec-http "4.1.77.Final"] - [io.netty/netty-handler "4.1.77.Final"] - [io.netty/netty-handler-proxy "4.1.77.Final"] - [io.netty/netty-resolver "4.1.77.Final"] - [io.netty/netty-resolver-dns "4.1.77.Final"] - - ;; Use specific commons-compress version to avoid - ;; CVE-2021-36090 pulled by avro 1.9.2 - [org.apache.commons/commons-compress "1.21"]] + [metosin/jsonista "0.3.7"]] :aliases {"kaocha" ["run" "-m" "kaocha.runner"]} :aot [jackdaw.serdes.edn2 jackdaw.serdes.fressian jackdaw.serdes.fn-impl] @@ -92,9 +77,9 @@ [org.apache.kafka/kafka-clients "3.2.0" :classifier "test"] [org.clojure/test.check "1.1.1"] [org.apache.kafka/kafka_2.13 "3.2.0"] - [lambdaisland/kaocha "1.66.1034"] - [lambdaisland/kaocha-cloverage "1.0.75"] - [lambdaisland/kaocha-junit-xml "0.0.76"]]} + [lambdaisland/kaocha "1.80.1274"] + [lambdaisland/kaocha-cloverage "1.1.89"] + [lambdaisland/kaocha-junit-xml "1.17.101"]]} ;; This is not in fact what lein defines repl to be :repl From b4f69ce613906931e305b03399189c5d3bdfa99c Mon Sep 17 00:00:00 2001 From: Chris McDevitt Date: Fri, 3 Mar 2023 18:06:14 +0000 Subject: [PATCH 3/4] deps: kafka streams 3.4.0 --- project.clj | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/project.clj b/project.clj index 99334f12..bd569186 100644 --- a/project.clj +++ b/project.clj @@ -19,9 +19,9 @@ :exclusions [com.fasterxml.jackson.core/jackson-databind]] [io.confluent/kafka-avro-serializer "7.3.2"] [io.confluent/kafka-json-schema-serializer "7.3.2"] - [org.apache.kafka/kafka-clients "3.2.0"] - [org.apache.kafka/kafka-streams "3.2.0"] - [org.apache.kafka/kafka-streams-test-utils "3.2.0"] + [org.apache.kafka/kafka-clients "3.4.0"] + [org.apache.kafka/kafka-streams "3.4.0"] + [org.apache.kafka/kafka-streams-test-utils "3.4.0"] [org.clojure/clojure "1.11.1" :scope "provided"] [org.clojure/java.data "1.0.95"] @@ -73,10 +73,10 @@ :resource-paths ["test/resources"] :injections [(require 'io.aviso.logging.setup)] :dependencies [[io.aviso/logging "1.0"] - [org.apache.kafka/kafka-streams-test-utils "3.2.0"] - [org.apache.kafka/kafka-clients "3.2.0" :classifier "test"] + [org.apache.kafka/kafka-streams-test-utils "3.4.0"] + [org.apache.kafka/kafka-clients "3.4.0" :classifier "test"] [org.clojure/test.check "1.1.1"] - [org.apache.kafka/kafka_2.13 "3.2.0"] + [org.apache.kafka/kafka_2.13 "3.4.0"] [lambdaisland/kaocha "1.80.1274"] [lambdaisland/kaocha-cloverage "1.1.89"] [lambdaisland/kaocha-junit-xml "1.17.101"]]} From 3eb81c67871d2658089d9d3f0cfa3337befbe196 Mon Sep 17 00:00:00 2001 From: Chris McDevitt Date: Fri, 31 Mar 2023 11:55:43 +0100 Subject: [PATCH 4/4] docs: update changelog --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3519e573..65b2d950 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ### Unreleased +- Make compatible with Kafka 3.4 + ### [0.9.9] - [2023-01-30] - Fix JSON_SR serde to correctly "Clojurize" deeply nested maps