            [java.util.concurrent TimeUnit]
            [java.util.concurrent.atomic AtomicLong]))
 
+
+
 (defn new-client ^AmazonS3Client [peer-config]
    (case (arg-or-default :onyx.peer/storage.s3.auth-type peer-config)
      :provider-chain (let [credentials (DefaultAWSCredentialsProviderChain.)]

        (recur (.listObjects client bucket prefix) new-ks)
        new-ks))))

(defrecord CheckpointManager [id monitoring client transfer-manager bucket transfers timeout-ns])

(defmethod onyx.checkpoint/storage :s3 [peer-config monitoring]
  (let [id (java.util.UUID/randomUUID)
        region (:onyx.peer/storage.s3.region peer-config)
        endpoint (:onyx.peer/storage.s3.endpoint peer-config)
        accelerate? (:onyx.peer/storage.s3.accelerate? peer-config)
        timeout-ns (ms->ns (arg-or-default :onyx.peer/storage.timeout peer-config))

        bucket (or (:onyx.peer/storage.s3.bucket peer-config)
                   (throw (Exception. ":onyx.peer/storage.s3.bucket must be supplied via peer-config when using :onyx.peer/storage = :s3.")))
        client (new-client peer-config)

      (.setMultipartCopyPartSize configuration (long v)))
    (when-let [v (:onyx.peer/storage.s3.multipart-upload-threshold peer-config)]
      (.setMultipartUploadThreshold configuration (long v)))
    (->CheckpointManager id monitoring client transfer-manager bucket encryption (atom []) timeout-ns)))

(defn checkpoint-task-key [tenancy-id job-id replica-version epoch task-id slot-id checkpoint-type]
  ;; We need to prefix the checkpoint key in a random way to partition keys for
