Lint all the code

.circleci/config.yml

@@ -100,6 +100,14 @@ jobs:
           key: v1-jackdaw-repo-{{ .Branch }}-{{ .Revision }}
           paths:
             - .
+  lint:
+    executor: machine
+    working_directory: /home/circleci/jackdaw
+    steps:
+      - checkout
+      - run: ls -la
+      - run: docker run --volume `pwd`:/project --rm --workdir /project cljkondo/clj-kondo sh -c 'clj-kondo --lint src test' || true
+
   deps:
     <<: *build_config
     steps:
@@ -111,6 +119,8 @@ jobs:
           key: *mvn_cache_key
           paths:
             - /home/circleci/.m2
+
+
   test:
     <<: *test_config
     steps:
@@ -159,6 +169,7 @@ workflows:
   version: 2
   build_and_test:
     jobs:
+      - lint
       - checkout_code
       - deps:
           requires:

.clj-kondo/config.edn

@@ -0,0 +1,7 @@
+{:linters {:unused-binding { ;; ignore unused :as binding.
+                            :exclude-destructured-as true}}
+
+ :lint-as {clojure.test.check.clojure-test/defspec clojure.core/def
+           jackdaw.data/defn->data clojure.core/defn
+           jackdaw.test.transports/deftransport clojure.core/defn
+           manifold.deferred/loop clojure.core/let}}

src/jackdaw/client.clj

@@ -26,16 +26,16 @@
 
 ;;;; Producer
 
-(defn ^KafkaProducer producer
+(defn producer
   "Return a producer with the supplied properties and optional Serdes."
-  ([config]
+  (^KafkaProducer [config]
    (KafkaProducer. ^java.util.Properties (jd/map->Properties config)))
-  ([config {:keys [^Serde key-serde ^Serde value-serde]}]
+  (^KafkaProducer [config {:keys [^Serde key-serde ^Serde value-serde]}]
    (KafkaProducer. ^java.util.Properties (jd/map->Properties config)
                    (.serializer key-serde)
                    (.serializer value-serde))))
 
-(defn ^Callback callback
+(defn callback
   "Return a kafka `Callback` function out of a clojure `fn`.
 
   The fn must be of 2-arity, being `[record-metadata?, ex?]` where the
@@ -44,7 +44,7 @@
   the record.
 
   Callbacks are `void`, so the return value is ignored."
-  [on-completion]
+  ^Callback [on-completion]
   (reify Callback
     (onCompletion [this record-meta exception]
       (on-completion record-meta exception))))
@@ -89,11 +89,11 @@
 
 ;;;; Consumer
 
-(defn ^KafkaConsumer consumer
+(defn consumer
   "Return a consumer with the supplied properties and optional Serdes."
-  ([config]
+  (^KafkaConsumer [config]
    (KafkaConsumer. ^java.util.Properties (jd/map->Properties config)))
-  ([config {:keys [^Serde key-serde ^Serde value-serde] :as t}]
+  (^KafkaConsumer [config {:keys [^Serde key-serde ^Serde value-serde] :as t}]
 
    (when-not (or key-serde
                  (get config "key.deserializer"))
@@ -134,15 +134,15 @@
                                 topic-configs))
   consumer)
 
-(defn ^KafkaConsumer subscribed-consumer
+(defn subscribed-consumer
   "Given a broker configuration and topics, returns a consumer that is
   subscribed to all of the given topic descriptors.
 
   WARNING: All topics subscribed to by a single consumer must share a
   single pair of key and value serde instances. The serdes of the
   first requested topic are used, and all other topics are expected to
   be able to use same serdes."
-  [config topic-configs]
+  ^KafkaConsumer [config topic-configs]
   (when-not (sequential? topic-configs)
     (throw (ex-info "subscribed-consumer takes a seq of topics!"
                     {:topic-configs topic-configs})))

src/jackdaw/client/partitioning.clj

@@ -34,7 +34,8 @@
   in `jackdaw.client` but backed by the partitioning machinery."
   {:license
    "BSD 3-Clause License <https://github.com/FundingCircle/jackdaw/blob/master/LICENSE>"}
-  (:require [jackdaw.client :as jc]
+  (:require [clojure.string :as str]
+            [jackdaw.client :as jc]
             [jackdaw.data :as jd])
   (:import org.apache.kafka.clients.producer.Producer
            org.apache.kafka.common.serialization.Serde
@@ -49,9 +50,9 @@
   [{:keys [record-key] :as t}]
   (let [record-key (as-> record-key %
                      (-> %
-                         (clojure.string/replace  "$." "")
-                         (clojure.string/replace  "_" "-")
-                         (clojure.string/split  #"\."))
+                         (str/replace  "$." "")
+                         (str/replace  "_" "-")
+                         (str/split  #"\."))
                      (mapv keyword %))]
     (assoc t ::key-fn #(get-in % record-key))))
 

src/jackdaw/data/admin.clj

@@ -16,7 +16,6 @@
     (ConfigEntry. k (:value v))))
 
 (defn->data ConfigEntry->data
-  ""
   [^ConfigEntry e]
   {:name (.name e)
    :value (.value e)
@@ -27,13 +26,11 @@
 ;;; Config
 
 (defn map->Config
-  ""
   ^Config [m]
   (Config.
    (map (partial apply ->ConfigEntry) m)))
 
 (defn->data Config->data
-  ""
   [^Config c]
   (into {}
         (comp (map ConfigEntry->data)
@@ -44,15 +41,13 @@
 ;;; TopicDescription
 
 (defn->data TopicDescription->data
-  ""
   [^TopicDescription td]
   {:is-internal? (.isInternal td)
    :partition-info (map datafy (.partitions td))})
 
 ;;; NewTopic
 
 (defn map->NewTopic
-  ""
   [{:keys [:topic-name
            :partition-count
            :replication-factor
@@ -71,7 +66,6 @@
 ;;;; Result types
 
 (defn->data DescribeClusterResult->data
-  ""
   [^DescribeClusterResult dcr]
   {:cluster-id (-> dcr .clusterId .get)
    :controller (-> dcr .controller .get datafy)

src/jackdaw/data/common.clj

@@ -10,7 +10,6 @@
 ;;; Node
 
 (defn->data Node->data
-  ""
   [^Node node]
   {:host (.host node)
    :port (.port node)
@@ -31,7 +30,6 @@
 ;;; TopicPartitionInfo
 
 (defn->data TopicPartitionInfo->data
-  ""
   [^TopicPartitionInfo tpi]
   {:isr (mapv datafy (.isr tpi))
    :leader (datafy (.leader tpi))
@@ -40,9 +38,9 @@
 
 ;;; Topic partition tuples
 
-(defn ^TopicPartition ->TopicPartition
+(defn ->TopicPartition
   "Given unrolled ctor-style arguments, create a Kafka `TopicPartition`."
-  [{:keys [:topic-name]} partition]
+  ^TopicPartition [{:keys [:topic-name]} partition]
   (TopicPartition. topic-name (int partition)))
 
 (defn map->TopicPartition
@@ -59,7 +57,6 @@
    :partition (.partition tp)})
 
 (defn as-TopicPartition
-  ""
   ^TopicPartition [o]
   (cond (instance? TopicPartition o)
         o

src/jackdaw/data/common_config.clj

@@ -8,15 +8,12 @@
 ;;; ConfigResource.Type
 
 (def +broker-config-resource-type+
-  ""
   ConfigResource$Type/BROKER)
 
 (def +topic-config-resource-type+
-  ""
   ConfigResource$Type/TOPIC)
 
 (def +unknown-config-resource-type+
-  ""
   ConfigResource$Type/UNKNOWN)
 
 (defn ->ConfigResourceType [o]
@@ -26,7 +23,6 @@
     +unknown-config-resource-type+))
 
 (defn->data ConfigResourceType->data
-  ""
   [^ConfigResource$Type crt]
   (cond (= +broker-config-resource-type+ crt)
         :config-resource/broker
@@ -40,22 +36,18 @@
 ;;; ConfigResource
 
 (defn ->ConfigResource
-  ""
   [^ConfigResource$Type type ^String name]
   (ConfigResource. type name))
 
 (defn ->topic-resource
-  ""
   [name]
   (->ConfigResource +topic-config-resource-type+ name))
 
 (defn ->broker-resource
-  ""
   [name]
   (->ConfigResource +broker-config-resource-type+ name))
 
 (defn->data ConfigResource->data
-  ""
   [^ConfigResource cr]
   {:name (.name cr)
    :type (datafy (.type cr))})

src/jackdaw/data/consumer.clj

@@ -6,15 +6,16 @@
 
 (import '[org.apache.kafka.clients.consumer
           ConsumerRecord OffsetAndTimestamp]
-        'org.apache.kafka.common.header.Headers)
+        'org.apache.kafka.common.header.Headers
+        'org.apache.kafka.common.record.TimestampType)
 
 (set! *warn-on-reflection* true)
 
-(defn ^ConsumerRecord ->ConsumerRecord
+(defn ->ConsumerRecord
   "Given unrolled ctor-style arguments create a Kafka `ConsumerRecord`.
 
   Convenient for testing the consumer API and its helpers."
-  [{:keys [:topic-name]} partition offset ts ts-type
+  ^ConsumerRecord [{:keys [:topic-name]} partition offset ts ts-type
    key-size value-size key value ^Headers headers]
   (ConsumerRecord. topic-name
                    (int partition)
@@ -72,8 +73,8 @@
 
 ;;; OffsetAndTimestamp tuples
 
-(defn ^OffsetAndTimestamp ->OffsetAndTimestamp
-  [{:keys [offset timestamp]}]
+(defn ->OffsetAndTimestamp
+  ^OffsetAndTimestamp [{:keys [offset timestamp]}]
   (OffsetAndTimestamp. offset (long timestamp)))
 
 (defn->data OffsetAndTimestamp->data [^OffsetAndTimestamp ots]

src/jackdaw/data/producer.clj

@@ -13,27 +13,27 @@
 
 ;;; Producer record
 
-(defn ^ProducerRecord ->ProducerRecord
+(defn ->ProducerRecord
   "Given unrolled ctor-style arguments creates a Kafka `ProducerRecord`."
-  ([{:keys [topic-name]} value]
+  (^ProducerRecord [{:keys [topic-name]} value]
    (ProducerRecord. ^String topic-name value))
-  ([{:keys [topic-name]} key value]
+  (^ProducerRecord [{:keys [topic-name]} key value]
    (ProducerRecord. ^String topic-name key value))
-  ([{:keys [topic-name]} partition key value]
-   (let [partition-or-nil (if partition (int partition))]
+  (^ProducerRecord [{:keys [topic-name]} partition key value]
+   (let [partition-or-nil (when partition (int partition))]
      (ProducerRecord. ^String topic-name
                       ^Integer partition-or-nil
                       key value)))
-  ([{:keys [topic-name]} partition timestamp key value]
-   (let [partition-or-nil (if partition (int partition))
-         timestamp-or-nil (if timestamp (long timestamp))]
+  (^ProducerRecord [{:keys [topic-name]} partition timestamp key value]
+   (let [partition-or-nil (when partition (int partition))
+         timestamp-or-nil (when timestamp (long timestamp))]
      (ProducerRecord. ^String topic-name
                       ^Integer partition-or-nil
                       ^Long timestamp-or-nil
                       key value)))
-  ([{:keys [topic-name]} partition timestamp key value headers]
-   (let [partition-or-nil (if partition (int partition))
-         timestamp-or-nil (if timestamp (long timestamp))]
+  (^ProducerRecord [{:keys [topic-name]} partition timestamp key value headers]
+   (let [partition-or-nil (when partition (int partition))
+         timestamp-or-nil (when timestamp (long timestamp))]
      (ProducerRecord. ^String topic-name
                       ^Integer partition-or-nil
                       ^Long timestamp-or-nil
@@ -87,26 +87,26 @@
                     offset 0 ;; Force absolute offset
                     timestamp
                     nil ;; No checksum, it's deprecated
-                    ^Integer (if key-size (int key-size))
-                    ^Integer (if value-size (int value-size))))
+                    ^Integer (when key-size (int key-size))
+                    ^Integer (when value-size (int value-size))))
   ([{:keys [:topic-name] :as t} partition base-offset relative-offset timestamp
     key-size value-size]
    (RecordMetadata. (->TopicPartition t partition)
                     base-offset
                     relative-offset ;; Full offset control
                     timestamp
                     nil ;; No checksum, it's depreciated
-                    ^Integer (if key-size (int key-size))
-                    ^Integer (if value-size (int value-size))))
+                    ^Integer (when key-size (int key-size))
+                    ^Integer (when value-size (int value-size))))
   ([{:keys [:topic-name] :as 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 (if key-size (int key-size))
-                    ^Integer (if value-size (int value-size)))))
+                    ^Integer (when key-size (int key-size))
+                    ^Integer (when value-size (int value-size)))))
 
 (defn map->RecordMetadata
   "Given a `::record-metdata`, build an equivalent `RecordMetadata`.

src/jackdaw/serdes/avro.clj

@@ -68,12 +68,12 @@
             KafkaAvroSerializer KafkaAvroDeserializer]
            java.lang.CharSequence
            java.nio.ByteBuffer
-           [java.io ByteArrayOutputStream ByteArrayInputStream]
+           [java.io ByteArrayOutputStream]
            [java.util Collection Map UUID]
            [org.apache.avro
             AvroTypeException Schema$Parser Schema$ArraySchema Schema Schema$Field]
            [org.apache.avro.io
-            EncoderFactory DecoderFactory JsonEncoder]
+            EncoderFactory DecoderFactory]
            [org.apache.avro.generic
             GenericDatumWriter GenericDatumReader
             GenericContainer GenericData$Array GenericData$EnumSymbol
@@ -91,10 +91,10 @@
      (when schema-str
        (.parse (Schema$Parser.) ^String schema-str)))))
 
-(defn- ^String mangle [^String n]
+(defn- mangle ^String [^String n]
   (str/replace n #"-" "_"))
 
-(defn- ^String unmangle [^String n]
+(defn- unmangle ^String [^String n]
   (str/replace n #"_" "-"))
 
 (defn- dispatch-on-type-fields

src/jackdaw/serdes/edn.clj

@@ -11,7 +11,6 @@
   (:require [clojure.edn]
             [jackdaw.serdes.fn :as jsfn])
   (:import java.nio.charset.StandardCharsets
-           org.apache.kafka.common.serialization.Serde
            org.apache.kafka.common.serialization.Serdes))
 
 (set! *warn-on-reflection* true)