Export files to Cloudflare R2 using Scala and the AWS SDK

Kevin van Zonneveld

Co-founder · Amsterdam, The Netherlands ·

Use the AWS SDK for Java v2 to export finished local files from a Scala application to a private Cloudflare R2 bucket. This example gives you one command-line program with synchronous, asynchronous, FS2 streaming, and parallel upload modes, plus a way to check the stored bytes.

Why choose Cloudflare R2?

R2’s S3-compatible API lets Scala code use the Java SDK’s request signing and HTTP clients. The example targets a bucket without a jurisdiction restriction at the standard account endpoint. Jurisdiction-specific buckets need the corresponding endpoint.

Set up your Scala project

This recipe targets existing Scala 2.13 applications using sbt 1.x. Its tested pins are Scala 2.13.18, sbt 1.13.0, Cats Effect 3.7.1, FS2 3.14.0, and AWS SDK 2.56.0. Scala 3 and sbt 2 are separate migration choices; these pins keep the example in the Scala 2.13 build family.

Use Bash, cURL, and a maintained OpenJDK 21 installation with java on your PATH. The tested JDK patch is 21.0.12.1; use your JDK distributor’s current security update within 21. You do not need a separately installed Scala compiler or sbt command. Run this from a writable directory to create a new project; the subshell leaves your current directory unchanged:

(
  set -eu
  command -v java >/dev/null
  command -v curl >/dev/null
  java -version
  mkdir r2-scala
  cd r2-scala
  mkdir -p project src/main/scala
  curl -fsSLo sbt-launch.jar https://repo.maven.apache.org/maven2/org/scala-sbt/sbt-launch/1.13.0/sbt-launch-1.13.0.jar
)

If creation fails because r2-scala exists, choose another parent directory. If the download fails, keep the project and retry just the cURL command from inside it. Save the following files in r2-scala. Use project/build.properties for the launcher pin:

sbt.version=1.13.0

Save build.sbt:

scalaVersion := "2.13.18"

libraryDependencies ++= Seq(
  "software.amazon.awssdk" % "s3" % "2.56.0",
  "software.amazon.awssdk" % "netty-nio-client" % "2.56.0",
  "software.amazon.awssdk" % "apache-client" % "2.56.0",
  "org.typelevel" %% "cats-effect" % "3.7.1",
  "co.fs2" %% "fs2-io" % "3.14.0",
  "co.fs2" %% "fs2-reactive-streams" % "3.14.0"
)

Compile / run / fork := true

Configure the AWS SDK client

Create a bucket and R2 S3 credentials scoped to that bucket, with object read and write permission. Supply R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY, and R2_BUCKET through your process environment before running the program. Use a secret manager or your terminal’s environment configuration; do not put credentials in the Scala file or command arguments. The bucket stays private and must already exist.

The clients use region auto, path-style addressing, and disabled chunked encoding, following Cloudflare’s Java configuration. WHEN_REQUIRED disables optional SDK request checksums. The program selects Apache for synchronous HTTP and Netty for asynchronous HTTP, and closes each client with Cats Effect Resource.

Basic synchronous upload

Choose sync for a straightforward blocking PUT. It reads the file through RequestBody.fromFile on Cats Effect’s blocking pool. Cancellation waits for the blocking call to return or time out; choose async or fs2 when you need cancellation of the underlying future.

Asynchronous upload with cats effect

Save this complete program as src/main/scala/R2Upload.scala. async uses the SDK’s file body; fs2 supplies a Reactive Streams publisher. Both wait through Cats Effect’s CompletableFuture bridge, which requests cancellation of the future when its fiber is canceled.

import cats.effect.{ExitCode, IO, IOApp, Resource}
import cats.effect.implicits._
import cats.syntax.all._
import fs2.interop.reactivestreams._
import fs2.io.file.{Files, Flags, Path}
import java.io.InputStream
import java.net.URI
import java.nio.ByteBuffer
import java.security.MessageDigest
import java.time.Duration
import java.util.Optional
import org.reactivestreams.Subscriber
import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
import software.amazon.awssdk.core.async.AsyncRequestBody
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.retries.StandardRetryStrategy
import software.amazon.awssdk.services.s3.{S3AsyncClient, S3Client, S3Configuration}
import software.amazon.awssdk.services.s3.model.{GetObjectRequest, PutObjectRequest, PutObjectResponse, S3Exception}

object R2Upload extends IOApp {
  final case class Config(endpoint: URI, accessKey: String, secretKey: String, bucket: String)
  final case class Task(path: Path, key: String)

  private val service = S3Configuration.builder()
    .pathStyleAccessEnabled(true)
    .chunkedEncodingEnabled(false)
    .build()

  private def overrides: ClientOverrideConfiguration =
    ClientOverrideConfiguration.builder()
      .apiCallTimeout(Duration.ofMinutes(2))
      .apiCallAttemptTimeout(Duration.ofSeconds(45))
      .retryStrategy(StandardRetryStrategy.builder().maxAttempts(3).build())
      .build()

  private def credentials(config: Config): StaticCredentialsProvider =
    StaticCredentialsProvider.create(
      AwsBasicCredentials.create(config.accessKey, config.secretKey)
    )

  def asyncClient(config: Config): Resource[IO, S3AsyncClient] =
    Resource.fromAutoCloseable(IO.blocking {
      S3AsyncClient.builder()
        .httpClientBuilder(NettyNioAsyncHttpClient.builder()
          .connectionTimeout(Duration.ofSeconds(10))
          .readTimeout(Duration.ofSeconds(30))
          .writeTimeout(Duration.ofSeconds(30)))
        .endpointOverride(config.endpoint)
        .credentialsProvider(credentials(config))
        .region(Region.of("auto"))
        .serviceConfiguration(service)
        .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
        .overrideConfiguration(overrides)
        .build()
    })

  def syncClient(config: Config): Resource[IO, S3Client] =
    Resource.fromAutoCloseable(IO.blocking {
      S3Client.builder()
        .httpClientBuilder(ApacheHttpClient.builder()
          .connectionTimeout(Duration.ofSeconds(10))
          .socketTimeout(Duration.ofSeconds(30)))
        .endpointOverride(config.endpoint)
        .credentialsProvider(credentials(config))
        .region(Region.of("auto"))
        .serviceConfiguration(service)
        .requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
        .overrideConfiguration(overrides)
        .build()
    })

  private def request(bucket: String, task: Task, size: Long): PutObjectRequest = {
    require(size >= 0 && size <= 5L * 1024 * 1024 * 1024,
      "Use multipart upload for files larger than 5 GiB")
    PutObjectRequest.builder()
      .bucket(bucket).key(task.key)
      .contentType("application/octet-stream")
      .contentLength(size)
      .build()
  }

  def uploadSync(client: S3Client, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      IO.blocking(client.putObject(
        request(bucket, task, size), RequestBody.fromFile(task.path.toNioPath)
      ))
    }

  def uploadAsync(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      IO.fromCompletableFuture(IO {
        client.putObject(request(bucket, task, size),
          AsyncRequestBody.fromFile(task.path.toNioPath))
      })
    }

  def uploadStream(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
    Files[IO].size(task.path).flatMap { size =>
      val buffers = Files[IO].readAll(task.path, 64 * 1024, Flags.Read)
        .chunks.map(chunk => ByteBuffer.wrap(chunk.toArray))

      buffers.toUnicastPublisher.use { publisher =>
        val body = new AsyncRequestBody {
          override def contentLength(): Optional[java.lang.Long] =
            Optional.of(java.lang.Long.valueOf(size))

          override def subscribe(subscriber: Subscriber[_ >: ByteBuffer]): Unit =
            publisher.subscribe(subscriber)
        }
        IO.fromCompletableFuture(IO {
          client.putObject(request(bucket, task, size), body)
        })
      }
    }

  def uploadMany(client: S3AsyncClient, bucket: String,
      tasks: List[Task]): IO[List[PutObjectResponse]] =
    tasks.parTraverseN(4)(task => uploadAsync(client, bucket, task))

  private def digest(input: InputStream): Array[Byte] = {
    val hash = MessageDigest.getInstance("SHA-256")
    val buffer = new Array[Byte](64 * 1024)
    var count = input.read(buffer)
    while (count != -1) {
      hash.update(buffer, 0, count)
      count = input.read(buffer)
    }
    hash.digest()
  }

  def verify(client: S3Client, bucket: String, task: Task): IO[Unit] = {
    val local = Resource.fromAutoCloseable(IO.blocking(
      java.nio.file.Files.newInputStream(task.path.toNioPath)))
    val remote = Resource.fromAutoCloseable(IO.blocking(client.getObject(
      GetObjectRequest.builder().bucket(bucket).key(task.key).build())))
    (local, remote).tupled.use { case (source, stored) =>
      IO.blocking {
        require(MessageDigest.isEqual(digest(source), digest(stored)), "Stored bytes differ")
      }
    }
  }

  override def run(args: List[String]): IO[ExitCode] = {
    val operation = IO {
      require(args.length >= 3 && args.length % 2 == 1,
        "Usage: R2Upload sync|async|fs2|parallel|verify file key [file key ...]")
      val mode = args.head
      require(Set("sync", "async", "fs2", "parallel", "verify").contains(mode), "Unknown mode")
      require(mode == "parallel" || args.length == 3, "Only parallel accepts multiple files")
      val account = sys.env("R2_ACCOUNT_ID")
      require(account.matches("[a-fA-F0-9]{32}"), "Invalid R2 account ID")
      val config = Config(
        URI.create(s"https://$account.r2.cloudflarestorage.com"),
        sys.env("R2_ACCESS_KEY_ID"), sys.env("R2_SECRET_ACCESS_KEY"), sys.env("R2_BUCKET")
      )
      val tasks = args.tail.grouped(2).map(pair => Task(Path(pair.head), pair(1))).toList
      require(tasks.forall(task =>
        task.path.toString.matches("[A-Za-z0-9_./-]+") &&
        task.key.matches("[A-Za-z0-9_./-]+") && !task.key.startsWith("-")),
        "Use ASCII paths and keys containing only letters, digits, _, ., /, and -")
      require(tasks.map(_.key).distinct.size == tasks.size, "Use distinct destination keys")
      (mode, config, tasks)
    }.flatMap { case (mode, config, tasks) =>
      val upload = if (mode == "verify")
        syncClient(config).use(client => verify(client, config.bucket, tasks.head))
      else if (mode == "sync")
        syncClient(config).use(client => uploadSync(client, config.bucket, tasks.head).void)
      else
        asyncClient(config).use { client =>
          mode match {
            case "fs2" => uploadStream(client, config.bucket, tasks.head).void
            case "parallel" => uploadMany(client, config.bucket, tasks).void
            case _ => uploadAsync(client, config.bucket, tasks.head).void
          }
        }
      upload *> IO.println(if (mode == "verify") "Stored bytes match." else "Upload completed.")
    }

    operation.as(ExitCode.Success).handleErrorWith {
      case error: S3Exception =>
        IO.println(s"Upload failed (${error.getClass.getSimpleName}, HTTP ${error.statusCode()}).").as(ExitCode.Error)
      case error =>
        IO.println(s"Upload failed (${error.getClass.getSimpleName}). Check arguments, environment, file access, and network.")
          .as(ExitCode.Error)
    }
  }
}

Open a terminal in r2-scala. These commands create a small input, compile the program, and upload it. The first compilation downloads dependencies. File and key arguments in this launcher are limited to ASCII letters, digits, _, ., /, and -; use ./ before a local name beginning with a hyphen. Destination keys must not start with - and must be distinct within a batch. Keep source files unchanged until all attempts finish.

(
  set -eu
  printf 'Hello, R2!\n' > hello.txt
  java -Dsbt.server.autostart=false -jar sbt-launch.jar compile
  java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload async ./hello.txt uploads/hello.txt"
)

Replace async with sync or fs2 for the other single-file modes. A successful PUT replaces any object at the same key, including when the input is empty. Use a fresh key if you need to keep the previous object.

After success, read the private object back and compare SHA-256 hashes of the complete local and stored contents:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload verify ./hello.txt uploads/hello.txt"

The verifier prints Stored bytes match. only when both hashes agree, and returns a nonzero status for a mismatch or failed read. A completed PUT alone confirms service acceptance; run the comparison before deleting an important original. Verification also needs the source to remain unchanged.

Handling responses and errors

Upload completed. appears only after every selected upload and the client scope finish. Service errors return a nonzero exit code and HTTP status. Local configuration, file, and network errors also return a nonzero exit code, without printing credentials or raw SDK responses.

A canceled upload does not become successful completion. Cancellation or a lost response can still leave a committed object at the destination: inspect it before retrying. This program does not roll back objects or delete remote data on failure.

Retry logic with exponential back-off

The SDK’s standard retry strategy handles retryable failures, with at most three attempts per PUT. The API call timeout is two minutes including retries; the attempt timeout is 45 seconds. Connection establishment has a 10-second limit. These are request limits, not a two-minute deadline for a whole parallel batch. The transports’ 30-second read limits sit below the attempt timeout; Netty also has a 30-second write limit.

A slow transfer can time out even below the file-size limit. Check the destination, then adjust the limits for your connection before retrying the unchanged file. Do not add an unrestricted retry loop. The FS2 publisher can rerun its file stream for a new subscription, so it also needs a stable, re-readable source.

Upload many files in parallel

The parallel mode runs at most four upload effects at once. It cancels sibling effects if one fails and returns a failing status. Completed objects remain; a batch is not a transaction. After creating a second local file named report.pdf, run:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload parallel ./hello.txt uploads/hello.txt ./report.pdf uploads/report.pdf"

The limit covers active uploads, not the in-memory argument list or every buffer inside Netty and the SDK. It does not split one file into parallel parts.

Stream large files using fs2

fs2 reads 64 KiB chunks and exposes them as ByteBuffer elements under downstream demand. The publisher’s resource remains open through completion, failure, or cancellation. This avoids loading the entire file into a byte array; it is not a fixed bound on total JVM or HTTP-client memory.

All upload modes use single PUTs and reject files larger than 5 GiB before sending a request. Streaming does not add multipart upload or resume. For larger files or resumable parts, use R2 multipart uploads.

Security best practices

Use bucket-scoped credentials and choose destination keys deliberately. This server-side Scala program needs no browser CORS configuration or public bucket URL. For files supplied by users, finish validation before exporting them to a location your application serves.

Common errors and their fixes

Result	Next check
HTTP failure	Verify the account endpoint, bucket name, and the token’s object permissions. A status alone does not identify which setting is wrong.
Local failure	Check all four environment variables, the argument restrictions, readable files, and the 5 GiB limit.
Timeout or connection failure	Check connectivity and the request limits, then inspect the destination before retrying.
Partial parallel batch	Inspect each requested key; successful siblings remain even though the overall command failed.

Wrap-up

Use the SDK’s file body for ordinary exports, and FS2 when your existing stream pipeline needs to supply the bytes. If the export follows a Transloadit processing job, the 🤖 /cloudflare/store Robot can write the results to R2 instead of running this program.