   (fn [output-collector stream partition key message]
     (kv-restore! input-topic message)))
 
+(defn- compile-global-kv-state-handler [{:keys [input-topic state] :as stream}]
+  (fn [output-collector stream partition key message]
+    (global-kv-restore! input-topic message)))
+
 (defn- compile-state-topic-config [{:keys [state-topic state input-topic] :as stream}]
   (cond
 


  (:output-topic-partition-fn stream)

)

      ;; other messages are processed by the normal pipeline, however here
      ;; we need to emit the state messages as well.

      ;; this ugly stuff works because *stores* is thread-local
      (let [kv-state (get *stores* stream)
            state    (var-get kv-state)
            [new-state rich-events] (pipeline state message)
            txlog     (kv/tx-log new-state)
