"putObjectFromStream" example fails on current version - S3 streaming uploads

[RichardBradley]

Describe the bug

Hi,

Both of the examples at https://docs.aws.amazon.com/code-library/latest/ug/s3_example_s3_Scenario_UploadStream_section.html
and the example at https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/javav2/example_code/s3/src/main/java/com/example/s3/async/PutObjectFromStreamAsync.java (which is the same as the first example at the former link) are failing on the current version of the SDK

The error given is:

Exception in thread "main" java.util.concurrent.CancellationException: subscription has been cancelled.
	at software.amazon.awssdk.utils.async.SimplePublisher.lambda$doProcessQueue$9(SimplePublisher.java:286)
	at software.amazon.awssdk.utils.async.SimplePublisher$FailureMessage.get(SimplePublisher.java:413)
	at software.amazon.awssdk.utils.async.SimplePublisher$FailureMessage.access$700(SimplePublisher.java:397)
	at software.amazon.awssdk.utils.async.SimplePublisher.doProcessQueue(SimplePublisher.java:260)
	at software.amazon.awssdk.utils.async.SimplePublisher.processEventQueue(SimplePublisher.java:224)
	at software.amazon.awssdk.utils.async.SimplePublisher.send(SimplePublisher.java:128)
	at software.amazon.awssdk.utils.async.InputStreamConsumingPublisher.doBlockingWrite(InputStreamConsumingPublisher.java:58)
	at software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody.writeInputStream(BlockingInputStreamAsyncRequestBody.java:95)
	at [my code]

I have tested this on windows and Mac with the following versions, same error:

Repro 1 versions:

Windows 11
"java --version" prints:
openjdk 17.0.14 2025-01-21 LTS
OpenJDK Runtime Environment Corretto-17.0.14.7.1 (build 17.0.14+7-LTS)
OpenJDK 64-Bit Server VM Corretto-17.0.14.7.1 (build 17.0.14+7-LTS, mixed mode, sharing)
SDK version is

        <dependency>
            <groupId>software.amazon.awssdk</groupId>
            <artifactId>s3</artifactId>
            <version>2.29.21</version>
        </dependency>
        <dependency>
            <groupId>software.amazon.awssdk.crt</groupId>
            <artifactId>aws-crt</artifactId>
            <version>0.33.11</version>
        </dependency>

Repro 2 versions:

Mac
openjdk 21.0.2 2024-01-16
OpenJDK Runtime Environment (build 21.0.2+13-58)
OpenJDK 64-Bit Server VM (build 21.0.2+13-58, mixed mode, sharing)
SDK: 2.29.44

Regression Issue

• Select this option if this issue appears to be a regression.

Expected Behavior

Expected the stream to be uploaded to S3

Current Behavior

The error given is:

Exception in thread "main" java.util.concurrent.CancellationException: subscription has been cancelled.
	at software.amazon.awssdk.utils.async.SimplePublisher.lambda$doProcessQueue$9(SimplePublisher.java:286)
	at software.amazon.awssdk.utils.async.SimplePublisher$FailureMessage.get(SimplePublisher.java:413)
	at software.amazon.awssdk.utils.async.SimplePublisher$FailureMessage.access$700(SimplePublisher.java:397)
	at software.amazon.awssdk.utils.async.SimplePublisher.doProcessQueue(SimplePublisher.java:260)
	at software.amazon.awssdk.utils.async.SimplePublisher.processEventQueue(SimplePublisher.java:224)
	at software.amazon.awssdk.utils.async.SimplePublisher.send(SimplePublisher.java:128)
	at software.amazon.awssdk.utils.async.InputStreamConsumingPublisher.doBlockingWrite(InputStreamConsumingPublisher.java:58)
	at software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody.writeInputStream(BlockingInputStreamAsyncRequestBody.java:95)
	at [my code]

Reproduction Steps

Run the example code at https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/javav2/example_code/s3/src/main/java/com/example/s3/async/PutObjectFromStreamAsync.java

Possible Solution

No response

Additional Information/Context

No response

AWS Java SDK version used

2.29.21

JDK version used

openjdk 17.0.14, openjdk 21.0.2 2024-01-16

Operating System and version

Windows 11, Mac

[bhoradc]

Hello @RichardBradley,

Thank your for reporting the issue. I am unable to reproduce the error using this code example.

Here's the sample code I tried running locally, but it uploaded the S3 object successfully.

Main.java

package org.example;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import java.io.ByteArrayInputStream;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

public class Main {
    public static void main(String[] args) {
        String bucketName = "xxxx-demo-bucket-" + UUID.randomUUID();
        String key = UUID.randomUUID().toString();
        AsyncExampleUtils.createBucket(bucketName);
        PutObjectResponse putObjectResponse = putObjectFromStream(AsyncExampleUtils.client, bucketName,key);
        System.out.println("Object {} etag: {}"+ key+ putObjectResponse.eTag());
        System.out.println("Object {} uploaded to bucket {}."+ key+ bucketName);
    }
    public static PutObjectResponse putObjectFromStream(S3AsyncClient s33CrtAsyncClient, String bucketName, String key) {
        BlockingInputStreamAsyncRequestBody body =
                AsyncRequestBody.forBlockingInputStream(null); // 'null' indicates a stream will be provided later.
        CompletableFuture<PutObjectResponse> responseFuture =
                s33CrtAsyncClient.putObject(r -> r.bucket(bucketName).key(key), body);
        String randomString = AsyncExampleUtils.randomString();
        System.out.println("random string to upload: {}: length={}"+ randomString+ randomString.length());
        body.writeInputStream(new ByteArrayInputStream(randomString.getBytes()));
        PutObjectResponse response = responseFuture.join();
        System.out.println("Object {} uploaded to bucket {}."+ key+ bucketName);
        return response;
    }
}

AsyncExampleUtils.java

package org.example;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.s3.S3AsyncClient;
public class AsyncExampleUtils {
    public static S3AsyncClient client = S3AsyncClient.crtCreate();
    private static final Logger logger = LoggerFactory.getLogger(AsyncExampleUtils.class);
    public static void createBucket(String bucketName) {
        client.createBucket(b -> b.bucket(bucketName)).join();
        client.waiter().waitUntilBucketExists(b -> b.bucket(bucketName)).join();
        logger.info("Bucket {} created.", bucketName);
    }
    public static String randomString() {
        int length = (int)(Math.random()*100);
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < length; i++) {
            sb.append((char) (Math.random() * 26 + 97));
        }
        return sb.toString();
    }
}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>org.example</groupId>
    <artifactId>V2_putObjectFromStream_5864</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<!--        <aws.sdk.version>2.18.30</aws.sdk.version>-->
<!--        <aws.sdk.version>2.25.18</aws.sdk.version>-->
<!--        <aws.sdk.version>2.25.68</aws.sdk.version>-->
<!--        <aws.sdk.version>2.29.21</aws.sdk.version>-->
        <aws.sdk.version>2.29.44</aws.sdk.version>
<!--        <aws.sdk.version>2.30.13</aws.sdk.version>-->
    </properties>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>software.amazon.awssdk</groupId>
                <artifactId>bom</artifactId>
<!--                <version>2.18.30</version>-->
<!--                <version>2.25.18</version>-->
<!--                <version>2.25.68</version>-->
<!--                <version>2.29.21</version>-->
                <version>2.29.44</version>
<!--                <version>2.30.13</version>-->
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-bom</artifactId>
                <version>2.19.0</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>software.amazon.awssdk</groupId>
            <artifactId>s3</artifactId>
        </dependency>
                <dependency>
                    <groupId>software.amazon.awssdk.crt</groupId>
                    <artifactId>aws-crt</artifactId>
                    <version>0.33.11</version>
                </dependency>
        <!-- The following 3 entries are needed for Log4j2 with SLF4J -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j2-impl</artifactId>
        </dependency>
    </dependencies>
</project>

Along with the latest CRT version (0.33.11), I also tested with the newer version of the SDK (2.30.13), is it the same behavior?

In your case, I am unsure what is causing the subscription to cancel while trying to write the input stream to S3. Would you be able to share the CRT debug logs for review? It can be enabled programmatically as shown below - reference.

import software.amazon.awssdk.crt.Log;

// Execute this statement before constructing the SDK service client.
Log.initLoggingToFile(Log.LogLevel.Trace, "<path to file>");

Regards,
Chaitanya

[RichardBradley]

Thanks for your reply

Thanks to your detailed repro code, I was able to isolate the issue

My code was using

S3AsyncClient.create()

which causes the error shown above.

Your code uses

S3AsyncClient.crtCreate()

which works.

I suppose this is largely user error then, as the detailed example code does create its client in that way.

I would note that

1: the example at https://docs.aws.amazon.com/code-library/latest/ug/s3_example_s3_Scenario_UploadStream_section.html does not show how to create the client, and gives no indication that the default client "create()" method will give this unhelpful error.

2: this error message is terrible and gives no indication as to what's wrong

3: the fact that "crtCreate()" and "create()" return the same Java class is a code contract that these two objects will be compatible, so it is very unhelpful API design to have the two differ in their capabilities

Please could you consider enhancing the API to make the "crt" and non-crt clients different classes, and to make the non-crt client throw a helpful error when trying to upload from a stream if this is not supported?

And please could you update the above linked docs to make this distinction clear?

Many thanks,

Rich