moves the upgrade-from config to stream-routes

gojek · Mar 25, 2019 · 523d60b · 523d60b
1 parent 323590a
commit 523d60b
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 5 deletions.
diff --git a/resources/config.test.ci.edn b/resources/config.test.ci.edn
@@ -33,6 +33,7 @@
                                              :stream-threads-count [1 :int]
                                              :origin-topic         "kafka-topic-*"
                                              :proto-class          "com.company.LogMessage"
+                                             :upgrade-from         "1.1"
                                              :channels             {:channel-1 {:worker-count [10 :int]
                                                                                 :retry        {:count   [5 :int]
                                                                                                :enabled [true :bool]}}}}}}}
diff --git a/resources/config.test.edn b/resources/config.test.edn
@@ -33,6 +33,7 @@
                                              :stream-threads-count [1 :int]
                                              :origin-topic         "kafka-topic-*"
                                              :proto-class          "com.company.LogMessage"
+                                             :upgrade-from         "1.1"
                                              :channels             {:channel-1 {:worker-count [10 :int]
                                                                                 :retry        {:count   [5 :int]
                                                                                                :enabled [true :bool]}}}}}}}
diff --git a/src/ziggurat/streams.clj b/src/ziggurat/streams.clj
@@ -31,16 +31,16 @@
   "Populates the upgrade.from config in kafka streams required for upgrading kafka-streams version from 1 to 2. If the
   value is non-nil it sets the config (the value validation is done in the kafka streams code), to unset the value the
   config needs to be set as nil "
-  [properties]
-  (if-let [upgrade-from-config (get-in config [:ziggurat :upgrade-from])]
+  [properties upgrade-from-config]
+  (if (some? upgrade-from-config)
     (.put properties StreamsConfig/UPGRADE_FROM_CONFIG upgrade-from-config)))
 
 (defn- validate-auto-offset-reset-config
   [auto-offset-reset-config]
   (if-not (contains? #{"latest" "earliest" nil} auto-offset-reset-config)
     (throw (ex-info "Stream offset can only be latest or earliest" {:offset auto-offset-reset-config}))))
 
-(defn- properties [{:keys [application-id bootstrap-servers stream-threads-count auto-offset-reset-config buffered-records-per-partition commit-interval-ms]}]
+(defn- properties [{:keys [application-id bootstrap-servers stream-threads-count auto-offset-reset-config buffered-records-per-partition commit-interval-ms upgrade-from]}]
   (validate-auto-offset-reset-config auto-offset-reset-config)
   (doto (Properties.)
     (.put StreamsConfig/APPLICATION_ID_CONFIG application-id)
@@ -52,7 +52,7 @@
     (.put StreamsConfig/BUFFERED_RECORDS_PER_PARTITION_CONFIG (int buffered-records-per-partition))
     (.put StreamsConfig/COMMIT_INTERVAL_MS_CONFIG commit-interval-ms)
     (.put ConsumerConfig/AUTO_OFFSET_RESET_CONFIG auto-offset-reset-config)
-    (set-upgrade-from-config)))
+    (set-upgrade-from-config upgrade-from)))
 
 (defn- get-metric-namespace [default topic]
   (str (name topic) "." default))

diff --git a/test/ziggurat/streams_test.clj b/test/ziggurat/streams_test.clj
@@ -14,7 +14,8 @@
 (def config-map {:stream-router {:vehicle {:application-id       "test"
                                            :bootstrap-servers    "localhost:9092"
                                            :stream-threads-count 1
-                                           :proto-class          "flatland.protobuf.test.Example$Photo"}}})
+                                           :proto-class          "flatland.protobuf.test.Example$Photo"
+                                           :upgrade-from         "1.1"}}})
 
 (def props (doto (Properties.)
              (.put ProducerConfig/BOOTSTRAP_SERVERS_CONFIG (get-in config-map [:stream-router :vehicle :bootstrap-servers]))