 (defmethod eval-op :flat-map-values
   [_ map-fn stream-or-table]
   `(.. ~(eval stream-or-table)
-       (flatMapValues (reify ValueMapper
-                        (apply [_ v#]
-                          (~map-fn v#))))))
+       (flatMapValues (value-mapper ~map-fn))))
 
 (defmethod eval-op :foreach
   [_ each-fn stream]

    (apply [_ k v]
      (apply key-value (map-fn k v)))))

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

(defmethod eval-op :map-values
  [_ map-fn stream]
  `(.. ~(eval stream)
       (mapValues (value-mapper ~map-fn))))

(defmethod eval-op :outer-join
  [_ join-fn left right & args]
