     (apply [_ k v]
       (apply key-value (map-fn k v)))))
 
+(defn value-mapper
+  [map-fn]
+  (reify ValueMapper
+    (apply [_ v]
+      (map-fn v))))
+
 ;; kstream/ktable ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;
 ;; Lots of methods exist on both KStream and KTable classes. Clojure

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

(defmethod eval-op :foreach
  [_ each-fn stream]

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