Add changelog-topic-replication to test

Co-authored-by: Kartik <rulz.gupta@gmail.com>
gojek · May 16, 2019 · d68d622 · d68d622
1 parent e8991e0
commit d68d622
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 4 deletions.
diff --git a/Makefile b/Makefile
@@ -4,6 +4,7 @@ all: test-all test
 topic="topic"
 
 setup:
+	docker-compose down
 	lein deps
 	docker-compose up -d
 	sleep 10

diff --git a/src/ziggurat/streams.clj b/src/ziggurat/streams.clj
@@ -40,7 +40,14 @@
   (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 upgrade-from]}]
+(defn- properties [{:keys [application-id
+                           bootstrap-servers
+                           stream-threads-count
+                           auto-offset-reset-config
+                           buffered-records-per-partition
+                           commit-interval-ms
+                           upgrade-from
+                           changelog-topic-replication-factor]}]
   (validate-auto-offset-reset-config auto-offset-reset-config)
   (doto (Properties.)
     (.put StreamsConfig/APPLICATION_ID_CONFIG application-id)

diff --git a/test/ziggurat/streams_test.clj b/test/ziggurat/streams_test.clj
@@ -8,7 +8,8 @@
            [java.util Properties]
            [kafka.utils MockTime]
            [org.apache.kafka.clients.producer ProducerConfig]
-           [org.apache.kafka.streams KeyValue]))
+           [org.apache.kafka.streams KeyValue]
+           [org.apache.kafka.streams.integration.utils IntegrationTestUtils]))
 
 (use-fixtures :once fix/mount-only-config)
 
@@ -42,11 +43,13 @@
                               :success)]
       (let [times                         6
             oldest-processed-message-in-s 10
+            changelog-topic-replication-factor 1
             kvs                           (repeat times message-key-value)
             streams                       (start-streams {:default {:handler-fn mapped-fn}}
                                                          (-> (ziggurat-config)
                                                              (assoc-in [:stream-router :default :application-id] (rand-application-id))
-                                                             (assoc-in [:stream-router :default :oldest-processed-message-in-s] oldest-processed-message-in-s)))]
+                                                             (assoc-in [:stream-router :default :oldest-processed-message-in-s] oldest-processed-message-in-s)
+                                                             (assoc-in [:stream-router :default :changelog-topic-replication-factor] changelog-topic-replication-factor)))]
         (Thread/sleep 10000)                                ;;waiting for streams to start
         (IntegrationTestUtils/produceKeyValuesSynchronously (get-in (ziggurat-config) [:stream-router :default :origin-topic])
                                                             kvs
@@ -63,10 +66,12 @@
                                 (swap! message-received-count inc))
                               :success)]
       (let [times   6
+            changelog-topic-replication-factor 1
             kvs     (repeat times message-key-value)
             streams (start-streams {:default {:handler-fn mapped-fn}}
                                    (-> (ziggurat-config)
-                                       (assoc-in [:stream-router :default :application-id] (rand-application-id))))]
+                                       (assoc-in [:stream-router :default :application-id] (rand-application-id))
+                                       (assoc-in [:stream-router :default :changelog-topic-replication-factor] changelog-topic-replication-factor)))]
         (Thread/sleep 10000)                                ;;waiting for streams to start
         (IntegrationTestUtils/produceKeyValuesSynchronously (get-in (ziggurat-config) [:stream-router :default :origin-topic])
                                                             kvs