How to upload multipart to Amazon S3 asynchronously using the java SDK

Question

In my java application I need to write data to S3, which I don't know the size in advance and sizes are usually big so as recommend in the AWS S3 documentation I am using the Using the Java AWS SDKs (low-level-level API) to write data to the s3 bucket.

In my application I provide S3BufferedOutputStream which is an implementation OutputStream where other classes in the app can use this stream to write to the s3 bucket.

I store the data in a buffer and loop and once the data is bigger than bucket size I upload data in the buffer as a a single UploadPartRequest Here is the implementation of the write method of S3BufferedOutputStream

@Override
public void write(byte[] b, int off, int len) throws IOException {
    this.assertOpen();
    int o = off, l = len;
    int size;
    while (l > (size = this.buf.length - position)) {
        System.arraycopy(b, o, this.buf, this.position, size);
        this.position += size;
        flushBufferAndRewind();
        o += size;
        l -= size;
    }
    System.arraycopy(b, o, this.buf, this.position, l);
    this.position += l;
}

The whole implementation is similar to this: code repo

My problem here is that each UploadPartRequest is done synchronously, so we have to wait for one part to be uploaded to be able to upload the next part. And because I am using the AWS S3 low level API I can not benefit from the parallel uploading provided by the TransferManager

Is there a way to achieve the parallel upload using low level SDK? Or some code changes that can be done to operate Asynchronously without corrupting the uploaded data and maintain order of the data?

kdgregory_42126 · Answer

Here's some example code from a class that I have. It submits the parts to an ExecutorService and holds onto the returned Future. This is written for the v1 Java SDK; if you're using the v2 SDK you could use an async client rather than the explicit threadpool:

// WARNING: data must not be updated by caller; make a defensive copy if needed
public synchronized void uploadPart(byte[] data, boolean isLastPart)
{
    partNumber++;
    logger.debug("submitting part {} for s3://{}/{}", partNumber, bucket, key);

    final UploadPartRequest request = new UploadPartRequest()
                                      .withBucketName(bucket)
                                      .withKey(key)
                                      .withUploadId(uploadId)
                                      .withPartNumber(partNumber)
                                      .withPartSize(data.length)
                                      .withInputStream(new ByteArrayInputStream(data))
                                      .withLastPart(isLastPart);

    futures.add(
        executor.submit(new Callable()
        {
            @Override
            public PartETag call() throws Exception
            {
                int localPartNumber = request.getPartNumber();
                logger.debug("uploading part {} for s3://{}/{}", localPartNumber, bucket, key);
                UploadPartResult response = client.uploadPart(request);
                String etag = response.getETag();
                logger.debug("uploaded part {} for s3://{}/{}; etag is {}", localPartNumber, bucket, key, etag);
                return new PartETag(localPartNumber, etag);
            }
        }));
}

Note: this method is synchronized to ensure that parts are not submitted out of order.

Once you've submitted all of the parts, you use this method to wait for them to finish and then complete the upload:

public void complete()
{
    logger.debug("waiting for upload tasks of s3://{}/{}", bucket, key);
    List partTags = new ArrayList<>();
    for (Future future : futures)
    {
        try
        {
            partTags.add(future.get());
        }
        catch (Exception e)
        {
            throw new RuntimeException(String.format("failed to complete upload task for s3://%s/%s"), e);
        }
    }

    logger.debug("completing multi-part upload for s3://{}/{}", bucket, key);
    CompleteMultipartUploadRequest request = new CompleteMultipartUploadRequest()
                                              .withBucketName(bucket)
                                              .withKey(key)
                                              .withUploadId(uploadId)
                                              .withPartETags(partTags);
    client.completeMultipartUpload(request);
    logger.debug("completed multi-part upload for s3://{}/{}", bucket, key);
}

You'll also need an abort() method that cancels outstanding parts and aborts the upload. This, and the rest of the class, are left as an exercise for the reader.

user_1435543 · Answer

You should look at using the AWS SDK for Java V2. You are referencing V1, not the newest Amazon S3 Java API. If you are not familiar with V2, start here:

Get started with the AWS SDK for Java 2.x

To perform Async operations via the Amazon S3 Java API, you use S3AsyncClient.

Now to learn how to upload an object using this client, see this code example:

import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import java.nio.file.Paths;
import java.util.concurrent.CompletableFuture;
// snippet-end:[s3.java2.async_ops.import]
// snippet-start:[s3.java2.async_ops.main]

/**
 * To run this AWS code example, ensure that you have setup your development environment, including your AWS credentials.
 *
 * For information, see this documentation topic:
 *
 * https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/get-started.html
 */

public class S3AsyncOps {

     public static void main(String[] args) {

         final String USAGE = "
" +
                 "Usage:
" +
                 "    S3AsyncOps   

" +
                 "Where:
" +
                 "    bucketName - the name of the Amazon S3 bucket (for example, bucket1). 

" +
                 "    key - the name of the object (for example, book.pdf). 
" +
                 "    path - the local path to the file (for example, C:/AWS/book.pdf). 
" ;

        if (args.length != 3) {
            System.out.println(USAGE);
             System.exit(1);
        }

        String bucketName = args[0];
        String key = args[1];
        String path = args[2];

        Region region = Region.US_WEST_2;
        S3AsyncClient client = S3AsyncClient.builder()
                .region(region)
                .build();

        PutObjectRequest objectRequest = PutObjectRequest.builder()
                .bucket(bucketName)
                .key(key)
                .build();

        // Put the object into the bucket
        CompletableFuture future = client.putObject(objectRequest,
                AsyncRequestBody.fromFile(Paths.get(path))
        );
        future.whenComplete((resp, err) -> {
            try {
                if (resp != null) {
                    System.out.println("Object uploaded. Details: " + resp);
                } else {
                    // Handle error
                    err.printStackTrace();
                }
            } finally {
                // Only close the client when you are completely done with it
                client.close();
            }
        });

        future.join();
    }
}

That is uploading an object using the S3AsyncClient client. To perform a multi-part upload, you need to use this method:

https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/s3/S3AsyncClient.html#createMultipartUpload-software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest-

TO see an example of Multipart upload using the S3 Sync client, see:

https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/javav2/example_code/s3/src/main/java/com/example/s3/S3ObjectOperations.java

That is your solution - use S3AsyncClient object's createMultipartUpload method.

How to upload multipart to Amazon S3 asynchronously using the java SDK

2 Answers