                             opts)
            :lifecycles [{:lifecycle/task task-name
                          :lifecycle/calls :onyx.plugin.kafka/read-messages-calls}]}
-    :schema {:task-map KafkaInputTaskMap
-             :lifecycles [os/Lifecycle]}})
+    :schema {:task-map KafkaInputTaskMap}})
   ([task-name :- s/Keyword
     topic :- s/Str
     group-id :- s/Str

                            opts)
           :lifecycles [{:lifecycle/task task-name
                         :lifecycle/calls :onyx.plugin.kafka/write-messages-calls}]}
    :schema {:task-map KafkaOutputTaskMap
             :lifecycles [os/Lifecycle]}})
  ([task-name :- s/Keyword
    topic :- s/Str
    zookeeper :- s/Str

  (.getBytes (pr-str segment)))

(def KafkaOutputTaskMap
  {:kafka/topic s/Str
   :kafka/zookeeper s/Str
   :kafka/serializer-fn os/NamespacedKeyword
   :kafka/request-size s/Num
   (s/optional-key :kafka/no-seal?) s/Bool
   (os/restricted-ns :kafka) s/Any})

(s/defn ^:always-validate producer
  ([task-name :- s/Keyword opts]
