              :lifecycles [os/Lifecycle]}})
   ([task-name :- s/Keyword
     topic :- s/Str
-    partition :- s/Str
     group-id :- s/Str
     zookeeper :- s/Str
     offset-reset :- (s/enum :smallest :largest)

    request-size :- s/Num
    task-opts :- {s/Any s/Any}]
   (consumer task-name (merge {:kafka/topic topic
                                  :kafka/zookeeper zookeeper
                                  :kafka/serializer-fn serializer-fn
                                  :kafka/request-size request-size}
                                 task-opts))))

    deserializer-fn :- os/NamespacedKeyword
    task-opts :- {s/Any s/Any}]
   (consumer task-name (merge {:kafka/topic topic
                               :kafka/group-id group-id
                               :kafka/zookeeper zookeeper
                               :kafka/offset-reset offset-reset
                               :kafka/force-reset? force-reset?
                               :kafka/deserializer-fn deserializer-fn}
                              task-opts))))

;;;; Writer task
(defn serialize-message-json [segment]
