    [clojure.string :as str])
   (:import
    (org.apache.kafka.streams KeyValue)
+   (org.apache.kafka.streams.processor Processor ProcessorSupplier)
    (org.apache.kafka.streams.kstream Predicate
                                      ForeachAction
                                      ValueMapper KeyValueMapper

    (apply [_ k v]
      (each-fn k v))))

;; kstream/ktable ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Lots of methods exist on both KStream and KTable classes. Clojure

(defmethod eval-op :process
  ([_ process-fn stream]
   `(.. ~(eval stream)
        (process (processor-supplier ~process-fn))))
  ([_ process-fn punctuate-fn stream]
   `(.. ~(eval stream)
        (process (processor-supplier ~process-fn ~punctuate-fn)))))
                         
(defmethod eval-op :select-key
  [_ map-fn stream]
