Add compression.type to allowed Producer props

factorhouse · May 27, 2022 · 894a036 · 894a036
1 parent dc9ca65
commit 894a036
Show file tree

Hide file tree

Showing 4 changed files with 19 additions and 9 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,12 @@
 # Change Log
 All notable changes to this project will be documented in this file. This change log follows the conventions of [keepachangelog.com](http://keepachangelog.com/).
 
+## [0.2.11] - 2022-05-27
+### Changed
+- Add `compression.type` to allowed Producer properties
+- Default Producer `compression.type` to `gzip`
+- Bump dependencies
+
 ## [0.2.10] - 2022-02-03
 ### Changed
 - Additional connection fields allowed for Producer properties 

diff --git a/project.clj b/project.clj
@@ -1,4 +1,4 @@
-(defproject io.operatr/kpow-streams-agent "0.2.10"
+(defproject io.operatr/kpow-streams-agent "0.2.11"
   :description "kPow's Kafka Streams monitoring agent"
   :url "https://github.com/operatr-io/kpow-streams-agent"
   :license {:name         "Apache-2.0 License"
@@ -21,10 +21,10 @@
                    [:roles
                     [:role "developer"]
                     [:role "maintainer"]]]])
-  :dependencies [[org.clojure/clojure "1.10.3"]
+  :dependencies [[org.clojure/clojure "1.11.1"]
                  [com.cognitect/transit-clj "1.0.329"]
                  [org.clojure/tools.logging "1.2.4"]
-                 [org.apache.kafka/kafka-streams "3.1.0" :scope "provided" :exclusions [com.fasterxml.jackson.core/jackson-core]]]
+                 [org.apache.kafka/kafka-streams "3.2.0" :scope "provided" :exclusions [com.fasterxml.jackson.core/jackson-core]]]
   :uberjar {:prep-tasks ["clean" "javac" "compile"]
             :aot        :all}
   :classifiers [["sources" {:source-paths      ^:replace []
@@ -33,13 +33,13 @@
                 ["javadoc" {:source-paths      ^:replace []
                             :java-source-paths ^:replace []
                             :resource-paths    ^:replace ["javadoc"]}]]
-  :profiles {:kaocha {:dependencies [[lambdaisland/kaocha "1.63.998"]]}
+  :profiles {:kaocha {:dependencies [[lambdaisland/kaocha "1.66.1034"]]}
              :dev    {:resource-paths ["dev-resources"]
                       :plugins        [[lein-cljfmt "0.7.0"]]
                       :dependencies   [[org.slf4j/slf4j-api "1.7.36"]
-                                       [ch.qos.logback/logback-classic "1.2.10"]
-                                       [cheshire "5.10.2" :exclusions [com.fasterxml.jackson.core/jackson-databind]]
-                                       [clj-kondo "2022.02.09"]]}
+                                       [ch.qos.logback/logback-classic "1.2.11"]
+                                       [cheshire "5.11.0" :exclusions [com.fasterxml.jackson.core/jackson-databind]]
+                                       [clj-kondo "2022.04.25"]]}
              :smoke  {:pedantic? :abort}}
   :aliases {"kaocha" ["with-profile" "+kaocha" "run" "-m" "kaocha.runner"]
             "kondo"  ["with-profile" "+smoke" "run" "-m" "clj-kondo.main" "--lint" "src"]

diff --git a/src/clojure/io/operatr/kpow/agent.clj b/src/clojure/io/operatr/kpow/agent.clj
@@ -152,8 +152,8 @@
         (metrics-send ctx (numeric-metrics metrics))
         ctx))))
 
-(defn ^Runnable snapshot-task
-  [snapshot-topic producer registered-topologies latch]
+(defn snapshot-task
+  ^Runnable [snapshot-topic producer registered-topologies latch]
   (fn []
     (let [job-id (str (UUID/randomUUID))
           ctx    {:job-id         job-id

diff --git a/src/java/io/operatr/kpow/StreamsRegistry.java b/src/java/io/operatr/kpow/StreamsRegistry.java
@@ -63,6 +63,10 @@ public static Properties filterProperties(Properties props) {
                 nextProps.setProperty(key, String.valueOf(props.get(key)));
             }
         }
+
+        String compressionType = props.getProperty("compression.type", "gzip");
+        nextProps.setProperty("compression.type", compressionType);
+
         return nextProps;
     }