 
       ;; flushing tx-log
       (let [new-state' (kv/flush-tx-log new-state txlog)]
-        (when-not (compare-and-set! *store* state new-state')
-          (throw (ex-info "Illegal state modification"
-                          {:new-state new-state ;; with tx-log
-                           :event     message})))))))
+        (var-set *store* new-state')))))
 
 
 

  "Takes an event in json format which contain an Tx-Log entry and
   it pushes the change to the running *kv-store*"
  [^String event]
  (swap! *store*
         (fn [store]
           (->> event
                from-json
                vector
                (kv/restore store)))))




  (alter-var-root #'*pipeline*     (constantly (core/make-samsara-processor config)))
  (alter-var-root #'*raw-pipeline* (constantly (make-raw-pipeline config)))
  (alter-var-root #'*config*       (constantly config))
  (alter-var-root #'*store*        (constantly (thread-local (kv/make-in-memory-kvstore))))
  (display-samza-config (:topics config)))


