 
             (= (Transfer$TransferState/Completed) state)
             (do
+             (info "Completed checkpoint to s3 under key" (:key @metric))
              (m/update-timer-ns! (:checkpoint-store-latency monitoring) 
                                  (- (System/nanoTime) (:start-time @metric)))
              (.addAndGet ^AtomicLong (:checkpoint-written-bytes monitoring) (:size-bytes @metric))

   task-id slot-id checkpoint-type ^bytes checkpoint-bytes]
  (let [k (checkpoint-task-key tenancy-id job-id replica-version epoch task-id
                               slot-id checkpoint-type)
        _ (info "Writing checkpoint to s3 under key" k)
        up ^Upload (onyx.storage.s3/upload ^TransferManager transfer-manager
                                           bucket
                                           k

                                           "application/octet-stream"
                                           :none)]
    (reset! upload up)
    (reset! metric {:key k
                    :size-bytes (alength checkpoint-bytes)
                    :start-time (System/nanoTime)})
    storage))

