   (let [tc (TransportConfiguration. (.getName NettyConnectorFactory))
         locator (HornetQClient/createServerLocatorWithoutHA (into-array [tc]))
         session-factory (.createSessionFactory locator)
-        session (.createSession session-factory)
+        session (.createTransactedSession session-factory)
         queue (:hornetq/queue-name task)
         producer (.createProducer session queue)]
     (.start session)

               (.acknowledge m)
               m)
          rets (doall (take-while (comp not nil?) (repeatedly (:hornetq/batch-size task) f)))]
      (.close consumer)
      (.close session)
      rets)))

(defn decompress-segment [segment]

        (.send producer message)))
    (.commit session)
    (.close producer)
    (.close session)
    (.close session-factory)
    (.close locator)))

(defn read-batch-shim [{:keys [catalog task]}]
  (let [task-map (planning/find-task catalog task)
