Tident Kafka Spout Clojure Attempting to call unbound fn

[Michael Ravits <mi...@gmail.com>]

Hello Everyone,

I'm a setting up a Kafka Spout in Clojure like that, and using it a Trident
topology:

(defn create-scheme []
  (reify backtype.storm.spout.MultiScheme
    (deserialize [this bytes]
           (avroclj.avro/deserialize bytes)))
    (getOutputFields [this]
      (Fields. ["a" "b" "c"]))))

    (set! (.scheme spout-config) (create-scheme))
    (TransactionalTridentKafkaSpout. spout-config)))

What I get in Storm's worker logs is this: RuntimeException:
java.lang.IllegalStateException: Attempting to call unbound fn:
#'avroclj.avro/deserialize

I am not an expert on the topic of ClassLoaders, but my guess is that the
thread which calls the deserialize method on the reified object has a
different ClassLoader than the one used when (create-scheme) was called.

So I found a workaround for this: I call (require 'myns.myschemenamespace)
at the beginning of the deserialize method like that:

    (deserialize [this bytes]
      (do
 (require 'myns.myschemenamespace)
            (avroclj.avro/deserialize bytes