Update to kafka streams 2.8.0 (confluent 6.1.1)

.circleci/config.yml

@@ -43,23 +43,24 @@ references:
   test_config: &test_config
     docker:
       - image: circleci/clojure:lein-2.9.1
-      - image: confluentinc/cp-zookeeper:5.3.1
+      - image: confluentinc/cp-zookeeper:6.1.1
         environment:
           ZOOKEEPER_CLIENT_PORT: 2181
-      - image: confluentinc/cp-kafka:5.3.1
+      - image: confluentinc/cp-kafka:6.1.1
         environment:
           KAFKA_BROKER_ID: 1
           KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
           KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
           KAFKA_ZOOKEEPER_CONNECT: localhost:2181
-      - image: confluentinc/cp-schema-registry:5.3.1
+      - image: confluentinc/cp-schema-registry:6.1.1
         environment:
           SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: localhost:2181
           SCHEMA_REGISTRY_HOST_NAME: localhost
           SCHEMA_REGISTRY_LISTENERS: http://localhost:8081
-      - image: confluentinc/cp-kafka-rest:5.3.1
+      - image: confluentinc/cp-kafka-rest:6.1.1
         environment:
           KAFKA_REST_ZOOKEEPER_CONNECT: localhost:2181
+          KAFKA_REST_BOOTSTRAP_SERVERS: localhost:9092
           KAFKA_REST_LISTENERS: http://0.0.0.0:8082
           KAFKA_REST_SCHEMA_REGISTRY: http://localhost:8081
           KAFKA_REST_HOST_NAME: localhost

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+* Update Kafka to 2.8.0 (confluent 6.1.1)
+
 ## [0.7.10] - [2021-04-14]
 * Bump netty related packages to latest version (related to changes in `0.7.8` which fixes CVEs).
 

project.clj

@@ -15,12 +15,12 @@
                  ;; 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 "5.3.1"
+                 [io.confluent/kafka-schema-registry-client "6.1.1"
                   :exclusions [com.fasterxml.jackson.core/jackson-databind]]
-                 [io.confluent/kafka-avro-serializer "5.3.1"]
-                 [org.apache.kafka/kafka-clients "2.3.1"]
-                 [org.apache.kafka/kafka-streams "2.3.1"]
-                 [org.apache.kafka/kafka-streams-test-utils "2.3.1"]
+                 [io.confluent/kafka-avro-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"]
 
                  [org.clojure/clojure "1.10.1" :scope "provided"]
                  [org.clojure/data.json "0.2.6"]
@@ -83,10 +83,10 @@
               :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.3.1"]
-                             [org.apache.kafka/kafka-clients "2.3.1" :classifier "test"]
+                             [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.11 "2.3.1"]
+                             [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"]]}

src/jackdaw/streams/lambdas.clj

@@ -150,20 +150,20 @@
   (process [_ key message]
     (processor-fn @context key message)))
 
-(defn processor [processor-fn]
+(defn processor
   "Packages up a Clojure fn as a kstream processor."
+  [processor-fn]
   (FnProcessor. (atom nil) processor-fn))
 
 (deftype FnProcessorSupplier [processor-supplier-fn]
   ProcessorSupplier
   (get [this]
-    processor-supplier-fn))
+    (processor processor-supplier-fn)))
 
 (defn processor-supplier
   "Packages up a Clojure fn in a kstream processor supplier."
   [processor-fn]
-  (let [fn-processor (processor processor-fn)]
-    (FnProcessorSupplier. fn-processor)))
+  (FnProcessorSupplier. processor-fn))
 
 (deftype FnTransformerSupplier [transformer-supplier-fn]
   TransformerSupplier

src/jackdaw/test/fixtures.clj

@@ -22,8 +22,8 @@
 (defn- new-topic
   [t]
   (doto (NewTopic. (:topic-name t)
-                   (:partition-count t)
-                   (:replication-factor t))
+                   (int (:partition-count t))
+                   (short (:replication-factor t)))
     (.configs (:config t))))
 
 (defn list-topics

test/jackdaw/serdes/avro_test.clj

@@ -655,8 +655,9 @@
     (is (= (String. (round-trip serde "bananas" (.getBytes "hello")))
            "hello"))
 
-    (is (thrown? java.lang.IllegalArgumentException
-                 (round-trip serde "bananas" {:hello 3})))))
+    (let [res (round-trip serde "bananas" {"hello" 3})]
+      (is (instance? java.util.HashMap res))
+      (is (= {(Utf8. "hello") 3} res)))))
 
 
 (deftest decoupled-reader-simple-string-schema