 
     (= state :global)
     {input-topic {:topic-filter (compile-partition-filter stream)
-                  :handler (compile-kv-state-handler stream)}}
+                  :handler (compile-global-kv-state-handler stream)}}
     ))
 
 

;;

;; runtime pipeline initialized by init-pipeline!
(def ^:dynamic *config*       nil)
(def ^:dynamic *dispatchers*  nil)
(def           *store*        nil) ;; thread local only





(defn init-pipeline! [config]
  (log/info "Initialize processors...")
  (alter-var-root #'*config*        (constantly config))
  (alter-var-root #'*stores*        (constantly (init-stores config)))
  (alter-var-root #'*global-stores* (constantly (atom {})))
  (alter-var-root #'*dispatchers*   (constantly (compile-config config))))



