        (decompress (:data (zk/data conn node)))))))
 
 (defmethod extensions/read-chunk [ZooKeeper :chunk]
-  [{:keys [conn opts prefix] :as log} kw id & _]
-  (clean-up-broken-connections
-   (fn []
-     (let [node (str (chunk-path prefix) "/" id "/chunk")]
+  [{:keys [conn opts prefix] :as log} kw id & {:keys [task-id]}]
+  (clean-up-broken-connections
+   (fn []
+     (let [node (str (chunk-path prefix) "/" task-id "/chunk-" id)]
+       (decompress (:data (zk/data conn node)))))))
+
+(defmethod extensions/read-chunk [ZooKeeper :chunk-index]
+  [{:keys [conn opts prefix] :as log} kw id & {:keys [task-id]}]
+  (clean-up-broken-connections
+   (fn []
+     (let [node (str (chunk-path prefix) "/" task-id "/index")]
        (decompress (:data (zk/data conn node)))))))
 
 (defmethod extensions/read-chunk [ZooKeeper :origin]

  [{:keys [conn opts prefix] :as log} kw chunk id]
  (clean-up-broken-connections
   (fn []
     (let [node (str (chunk-path prefix) "/" id "/chunk")
           bytes (compress chunk)]
       (zk/create-all conn node :persistent? true :data bytes)
       id))))

(defmethod extensions/write-chunk [ZooKeeper :job-scheduler]
  [{:keys [conn opts prefix] :as log} kw chunk id]

  [{:keys [conn opts prefix] :as log} kw chunk id]
  (clean-up-broken-connections
   (fn []
     (let [node (str (chunk-path prefix) "/" id "/chunk-" (:id chunk))
           version (:version (zk/exists conn node))
           bytes (compress chunk)]
       (zk/set-data conn node bytes version)))))
