             [onyx.api])
   (:import [franzy.clients.producer.types ProducerRecord]))
 
+(def n-partitions 8)
 (defn build-job [zk-address topic batch-size batch-timeout]
   (let [batch-settings {:onyx/batch-size batch-size :onyx/batch-timeout batch-timeout}
         base-job (merge {:workflow [[:read-messages :identity]

                                    :kafka/offset-reset :earliest
                                    :kafka/force-reset? true
                                    :kafka/deserializer-fn :onyx.tasks.kafka/deserialize-message-edn
                                    :onyx/min-peers 2
                                    :onyx/max-peers 2}
                                   batch-settings)))
        (add-task (core-async/output :out batch-settings)))))


                                          :controlled.shutdown.enable false}))

        zk-utils (k-admin/make-zk-utils {:servers [zookeeper]} false)
        _ (k-topics/create-topic! zk-utils topic n-partitions)

        producer-config {:bootstrap.servers ["127.0.0.1:9092"]}
        key-serializer (byte-array-serializer)
