            bytes (compress chunk)]
        (zk/create conn node :persistent? true :data bytes)))))
 
-(defmethod extensions/write-chunk [ZooKeeper :bootstrapped-segment]
-  [{:keys [conn opts prefix] :as log} kw chunk id]
-  (clean-up-broken-connections
-   (fn []
-     (let [node (str (bootstrap-path prefix) "/" id "/segment-" (:id chunk))
+(defmethod extensions/write-chunk [ZooKeeper :chunk]
+  [{:keys [conn opts prefix] :as log} kw chunk id]
+  (clean-up-broken-connections
+   (fn []
+     (let [id (java.util.UUID/randomUUID)
+           node (str (chunk-path prefix) "/" id "/chunk-" id)
+           bytes (compress chunk)]
+       (zk/create-all conn node :persistent? true :data bytes)
+       id))))
+
+(defmethod extensions/write-chunk [ZooKeeper :chunk-index]
+  [{:keys [conn opts prefix] :as log} kw chunk id]
+  (clean-up-broken-connections
+   (fn []
+     (let [node (str (chunk-path prefix) "/" id "/index")
            bytes (compress chunk)]
        (zk/create-all conn node :persistent? true :data bytes)))))
 
-(defmethod extensions/write-chunk [ZooKeeper :bootstrapped-index]
-  [{:keys [conn opts prefix] :as log} kw chunk id]
-  (clean-up-broken-connections
-   (fn []
-     (let [node (str (bootstrap-path prefix) "/" id "/index")
-           bytes (compress chunk)]
-       (zk/create-all conn node :persistent? true :data bytes)))))
-
 (defmethod extensions/write-chunk [ZooKeeper :job-scheduler]
   [{:keys [conn opts prefix] :as log} kw chunk id]
   (clean-up-broken-connections

     (let [node (str (sentinel-path prefix) "/" id)]
       (decompress (:data (zk/data conn node)))))))

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

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

(defmethod extensions/read-chunk [ZooKeeper :origin]

(defn sentinel-path [prefix]
  (str (prefix-path prefix) "/sentinel"))

(defn chunk-path [prefix]
  (str (prefix-path prefix) "/chunk"))

(defn origin-path [prefix]
  (str (prefix-path prefix) "/origin"))
