                             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

    (catch Exception e
      {:error e})))

(def UserTaskMapKey
  (os/build-allowed-key-ns :kafka))

(def KafkaInputTaskMap
  (s/->Both [os/TaskMap
             {:kafka/topic s/Str
              :kafka/group-id s/Str
              :kafka/zookeeper s/Str
              :kafka/offset-reset (s/enum :smallest :largest)
              :kafka/force-reset? s/Bool
              :kafka/deserializer-fn os/NamespacedKeyword
              (s/optional-key :kafka/partition) s/Str
              (s/optional-key :kafka/chan-capacity) s/Num
              (s/optional-key :kafka/fetch-size) s/Num
              (s/optional-key :kafka/empty-read-back-off) s/Num
              (s/optional-key :kafka/commit-interval) s/Num
              (s/optional-key :kafka/wrap-with-metadata?) s/Bool
              UserTaskMapKey s/Any}]))

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

  (.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]
