 (defn ingest-file
   [path graph continuation]
   (doseq [file (kafka/dir->files path)]
+    (log/info "ingesting" (.getName file))
     (let [label (kafka/path->label (.getName file))
           lines (line-seq (io/reader file))]
       (doseq [line lines]


(defn ingest-vertex
  [graph data]
  (let [vertex (query/add-vertex! graph data)
        id (.id vertex)]
    (merge


(defn ingest-edge
  [graph data]
  (log/info "edge" (:gid data))
  (let [edge (query/add-edge! graph data)
        id (.id edge)
        from-id (.getOutVertexId id)
        type-id (.getTypeId id)
        edge-id (.getRelationId id)
        to-id (.getInVertexId id)]
    (log/info "ingesting" (.getName file))
    (merge
     (lift-properties data)
     {:_janusId edge-id
