store updated project

Author: source-c

README.adoc

@@ -4,15 +4,14 @@ A Clojure library for the Apache Kafka (distributed stream-processing software p
 
 Uses KF protocol and does not rely on ZooKeeper.
 
-NOTE: Still contains zookeeper as a dependency.
-
 Tries to be as lightweigh as possible thus depends only on
 
-- `org.apache.kafka/kafka_2.12 "3.4.0"`
-- `org.apache.kafka/kafka-clients "3.4.0"`
-- `org.apache.zookeeper/zookeeper "3.8.1"`
+- `org.apache.kafka/kafka_2.12 "3.9.0"`
+- `org.apache.kafka/kafka-clients "4.0.0"`
+
+but excluding `jmx*` and logging.
 
-but excluding `jms`,`jmx*` and logging.
+NOTE: The Zookeeper dependency has been removed as it's not required by modern Kafka clients for standard operations.
 
 NOTE: Some builds (for instance of `v0.4.x` branch) may partially (sometimes even fully) be incompatible with some versions of other libraries that also use NIO! If you’re experiencing build problems and/or your application is unexpectedly crashed on start - try check your project dependencies more deeply, may be you will need to correct existing dependencies version or to add an actual version of full `[io.netty/netty-all]`
 
@@ -31,12 +30,12 @@ image:https://img.shields.io/github/last-commit/source-c/clj-kafka-x?style=for-t
 
 Add the following to your http://github.com/technomancy/leiningen[Leiningen's]
 `project.clj`:
-
+ 
 [source,clojure]
 ----
-[net.tbt-post/clj-kafka-x "0.7.5"]
+[net.tbt-post/clj-kafka-x "0.8.0"]
 ----
-
+ 
 == Usage
 
 === Producer
@@ -124,7 +123,7 @@ $ lein install
 
 == License
 
-Copyright © 2016-2023
+Copyright © 2016-2025
 
 Distributed under the
 http://www.apache.org/licenses/LICENSE-2.0[Apache License v 2.0]

profiles.clj

@@ -1,5 +1,5 @@
-{:dev      {:dependencies [[midje "1.10.9"]]}
+{:dev      {:dependencies [[midje "1.10.10"]]}
  :docs     {:plugins [[lein-codox "0.10.8"]
                       [org.timmc/nephila "0.3.0"]]}
- :provided {:dependencies [[org.clojure/clojure "1.11.1"]]}
+ :provided {:dependencies [[org.clojure/clojure "1.12.0"]]}
  :jar      {:aot :all}}

project.clj

@@ -1,13 +1,10 @@
-(defproject net.tbt-post/clj-kafka-x "0.7.6"
+(defproject net.tbt-post/clj-kafka-x "0.8.0"
   :description "A Clojure wrapper for Apache Kafka v2/v3 client"
   :url "https://github.com/source-c/clj-kafka-x"
   :license {:name "Apache License 2.0"
             :url  "http://www.apache.org/licenses/LICENSE-2.0"}
-  :dependencies [[org.apache.kafka/kafka_2.12 "3.5.0"
-                  :exclusions [javax.jms/jms
-                               com.sun.jdmk/jmxtools
+  :dependencies [[org.apache.kafka/kafka_2.12 "3.9.0"
+                  :exclusions [com.sun.jdmk/jmxtools
                                com.sun.jmx/jmxri
                                org.slf4j/slf4j-log4j12]]
-                 [org.apache.zookeeper/zookeeper "3.8.1"
-                  :exclusions [org.slf4j/slf4j-log4j12]]
-                 [org.apache.kafka/kafka-clients "3.5.0"]])
+                 [org.apache.kafka/kafka-clients "4.0.0"]])

src/clj_kafka_x/consumers/simple.clj

@@ -1,6 +1,6 @@
 (ns ^{:doc "Clojure interface for Kafka Consumer API.
             For complete JavaDocs, see:
-            https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/package-summary.html"}
+            https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/package-summary.html"}
   clj-kafka-x.consumers.simple
   (:require [clj-kafka-x.data :refer :all]
             [clj-kafka-x.impl.helpers :refer :all])
@@ -20,12 +20,12 @@
   "Takes a map of config options and returns a `KafkaConsumer` for consuming records from Kafka.
 
   NOTE `KafkaConsumer` instances are NOT thread-safe, see
-  https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#multithreaded
-
+  https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#multithreaded
+ 
   For more information and available config options,
-  see: https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
+  see: https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
        http://kafka.apache.org/documentation.html#newconsumerconfigs
-
+ 
   Usage:
 
 ;; Created using just a map of configs, in this case the keys
@@ -88,12 +88,12 @@
                        {:topic \"topic-b\" :partitions #{0 1}}
                        {:topic \"topic-c\" :partitions #{0}}])
   ;; => nil
-
+ 
   For more in-depth information
-  https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)
-  http://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)
-  http://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)
-  https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)
+  https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection)
+  http://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.Collection,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)
+  http://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#subscribe(java.util.regex.Pattern,%20org.apache.kafka.clients.consumer.ConsumerRebalanceListener)
+  https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#assign(java.util.Collection)
   "
   [^Consumer consumer topics & {:keys [assigned-callback revoked-callback]
                                 :or   {assigned-callback (fn [_])
@@ -279,13 +279,13 @@
   ([^Consumer consumer offset-commit-fn]
    (let [callback (reify OffsetCommitCallback
                     (onComplete [_ offsets exception]
-                      (offset-commit-fn (tp-om-map->map offsets) exception)))]
+                      (offset-commit-fn (topic-partition-offsets->clj offsets) exception)))]
      (.commitAsync consumer callback)))
   ([^Consumer consumer topic-partition-offsets-metadata offset-commit-fn]
    (let [callback (reify OffsetCommitCallback
                     (onComplete [_ offsets exception]
-                      (offset-commit-fn (tp-om-map->map offsets) exception)))
-         tp-om-map (map->tp-om-map topic-partition-offsets-metadata)]
+                      (offset-commit-fn (topic-partition-offsets->clj offsets) exception)))
+         tp-om-map (clj->topic-partition-offsets-map topic-partition-offsets-metadata)]
      (.commitAsync consumer tp-om-map callback))))
 
 
@@ -311,15 +311,15 @@
   "
   ([^Consumer consumer] (.commitSync consumer))
   ([^Consumer consumer topic-partitions-offsets-metadata]
-   (let [tp-om-map ^Map (map->tp-om-map topic-partitions-offsets-metadata)]
+   (let [tp-om-map ^Map (clj->topic-partition-offsets-map topic-partitions-offsets-metadata)]
      (.commitSync consumer tp-om-map))))
 
 
 (defn ^:deprecated last-committed-offset
   "Gets the last committed offset for the partition of a topic.
    NOTE This function is a blocking I/O operation.
 
-   see http://kafka.apache.org/0100/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#committed(org.apache.kafka.common.TopicPartition)
+   see http://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#committed(org.apache.kafka.common.TopicPartition)
 
   Usage:
 
@@ -336,11 +336,11 @@
 (defn list-all-topics
   "Get metadata about ALL partitions for ALL topics that the user is authorized to view.
    NOTE This function is a blocking I/O operation.
-
-   See https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#listTopics()
-
+ 
+   See https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#listTopics()
+ 
   Usage :
-
+ 
   (list-all-topics consumer)
   ;; =>{\"topic-a\"
   ;;    [{:topic \"topic-a\",
@@ -366,16 +366,16 @@
   ;;      :in-sync-replicas [{:id 2, :host \"172.17.0.3\", :port 9093}]}]}
   "
   [^Consumer consumer]
-  (str-pi-map->map (.listTopics consumer)))
+  (topic-partitions-info->clj (.listTopics consumer)))
 
 (defn list-all-partitions
   "Get metadata about all partitions for a particular topic.
    NOTE This function is a blocking I/O operation.
-
-   See https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#partitionsFor(java.lang.String)
-
+ 
+   See https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#partitionsFor(java.lang.String)
+ 
   Usage :
-
+ 
   (list-all-partitions consumer)
   ;; => [{:topic \"topic-b\",
   ;;      :partition 2,
@@ -400,10 +400,10 @@
 (defn pause
   "Stops messages being consumed from the given partitions.
    This takes effect on the next call on the messages function
-   See https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)
-
+   See https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#pause(java.util.Collection)
+ 
   Usage:
-
+ 
   (pause consumer {:topic \"topic-a\" :partition 2}
                   {:topic \"topic-b\" :partition 0})
   "
@@ -416,10 +416,10 @@
 (defn resume
   "Resumes messages being consumed from the given partitions.
    This takes effect on the next call on the messages function
-   See https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#resume(java.util.Collection)
-
+   See https://kafka.apache.org/40/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#resume(java.util.Collection)
+ 
   Usage:
-
+ 
   (resume consumer {:topic \"topic-a\" :partition 2}
                    {:topic \"topic-b\" :partition 0})
   "

src/clj_kafka_x/data.clj

@@ -85,9 +85,8 @@
   (if (or (nil? offset) (nil? metadata))
     (throw (ex-info "Provided map is missing offset or metadata keys" m))
     (OffsetAndMetadata. offset metadata)))
-
-;;NOT happy with this function name
-(defn tp-om-map->map
+ 
+(defn topic-partition-offsets->clj
   "Takes a java.util.Map made of TopicPartition as keys and OffsetAndMetadata as values,
    converts them to the following clojure equivalent data structure
 
@@ -96,32 +95,30 @@
    {:topic \"dev\", :partition 1} {:offset 234, :metadata \"loads of data\"},
    {:topic \"dev\", :partition 7} {:offset 23, :metadata \"mega data\"}}
   "
-  [^Map tp-om]
+  [^Map topic-partition-offsets-map]
   (let [reduce-fn (fn [m [^TopicPartition tp ^OffsetAndMetadata om]]
                     (assoc m (to-clojure tp) (to-clojure om)))]
-    (reduce reduce-fn {} tp-om)))
-
-;;NOT happy with this function name
-(defn map->tp-om-map
+    (reduce reduce-fn {} topic-partition-offsets-map)))
+ 
+(defn clj->topic-partition-offsets-map
   "Takes a Clojure map (see below for example) and converts it to a java.util.Map made of TopicPartition as keys and OffsetAndMetadata as values
-
+ 
   {{:topic \"test\", :partition 77} {:offset 34, :metadata \"data data\"},
    {:topic \"prod\", :partition 4} {:offset 24, :metadata \"more data\"},
    {:topic \"dev\", :partition 1} {:offset 234, :metadata \"loads of data\"},
    {:topic \"dev\", :partition 7} {:offset 23, :metadata \"mega data\"}}
   "
   [m]
-  (let [tp-om-map (HashMap.)
+  (let [^Map java-map (HashMap.)
         reduce-fn (fn [^Map m kv]
                     (.put m (map->topic-partition (first kv))
                           (map->offset-metadata (second kv)))
                     m)]
-    (reduce reduce-fn tp-om-map  m)))
-
-
-;;NOT happy with this function name
-(defn str-pi-map->map
-  "Takes a java.util.Map made of Strings as keys and java.util.List <PartitionInfo> as values,
+    (reduce reduce-fn java-map m)))
+ 
+ 
+(defn topic-partitions-info->clj
+  "Takes a java.util.Map made of Strings (topic names) as keys and java.util.List <PartitionInfo> as values,
    converts it to a Map with the keys being the topic name and the values being a vector of maps (each map representing information about a topic partition)
   e.g
 
@@ -147,10 +144,10 @@
    :in-sync-replicas [{:id 3, :host \"172.17.0.4\", :port 9094} {:id 1, :host \"172.17.0.3\", :port 9092} {:id 2, :host \"172.17.0.2\", :port 9093}]}]}
   "
   [^Map str-pi]
-  (let [reduce-fn (fn [m [name pi-list]]
-                    (assoc m name (mapv to-clojure pi-list)))]
+  (let [reduce-fn (fn [clj-map [name pi-list]]
+                    (assoc clj-map name (mapv to-clojure pi-list)))]
     (reduce reduce-fn {} str-pi)))
-
+ 
 (defn metrics->map
   "Returns a sequence of maps, with each map representing a metric.
    The composition of each map is :group :name :description :tags :value

src/clj_kafka_x/producer.clj

@@ -1,6 +1,6 @@
 (ns ^{:doc "Clojure interface for Kafka Producer API. For
   complete JavaDocs, see:
-  https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/package-summary.html"}
+  https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/package-summary.html"}
   clj-kafka-x.producer
   (:refer-clojure :exclude [send flush])
   (:require [clj-kafka-x.data :refer :all]
@@ -30,12 +30,12 @@
   "Takes a map of config options and returns a `KafkaProducer` for publishing records to Kafka.
 
   NOTE `KafkaProducer` instances are thread-safe and should generally be shared for best performance.
-
+ 
   For more information and available config options,
-  see: https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html
+  see: https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html
        https://kafka.apache.org/documentation.html#producerconfigs
-
-
+ 
+ 
   Usage:
 
 ;; Created using just a map of configs, in this case the keys
@@ -74,11 +74,11 @@
   a callback fn that will be called when the operation completes.
   Callback should be a fn of two arguments, a map as above, and an
   exception. Exception will be nil if operation succeeded.
-
-  See: https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#send(org.apache.kafka.clients.producer.ProducerRecord)
-
+ 
+  See: https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#send(org.apache.kafka.clients.producer.ProducerRecord)
+ 
   Usage:
-
+ 
   ;;To send the message asynchronously and return a Future
   (send producer (record \"topic-a\" \"Test message 1\"))
   ;; => #object[string representation of future object]
@@ -101,19 +101,19 @@
                                       (onCompletion [_ metadata exception]
                                         (callback (and metadata (to-clojure metadata)) exception))))]
      (map-future-val fut to-clojure))))
-
+ 
 (defn flush
-  "See: https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#flush()"
+  "See: https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#flush()"
   [^Producer producer]
   (.flush producer))
 
 (defn close
   "Like `.close`, but with a default time unit of ms for the arity with timeout.
-
+ 
   See:
-
-  - https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#close()
-  - https://kafka.apache.org/34/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#close(java.time.Duration)"
+ 
+  - https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#close()
+  - https://kafka.apache.org/37/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#close(java.time.Duration)"
   ([^Producer producer]
    (.close producer))
   ([^Producer producer timeout-ms]