S3TransferManager: The TransferListener.transferComplete() is not invoked when AsyncRequestBody.fromFile() is used

[artembilan]

Describe the bug

The FileAsyncRequestBody does not call subscriber.onComplete() because it is cancelled from the NettyRequestExecutor.StreamedRequest via this code:

                        if (!shouldContinuePublishing()) {
                            done = true;
                            subscription.cancel();
                            subscriber.onComplete();
                        }

Expected Behavior

The FileAsyncRequestBody is properly completed and therefore a TransferListener.transferComplete() is signaled.

Current Behavior

The TransferListener.transferComplete() is not called when we upload the file.

Works for plain byte[] though.

Reproduction Steps

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.Path;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.testcontainers.containers.localstack.LocalStackContainer;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.core.ResponseBytes;
import software.amazon.awssdk.core.async.AsyncRequestBody;
import software.amazon.awssdk.core.async.AsyncResponseTransformer;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.transfer.s3.S3TransferManager;
import software.amazon.awssdk.transfer.s3.model.UploadRequest;
import software.amazon.awssdk.transfer.s3.progress.LoggingTransferListener;
import software.amazon.awssdk.transfer.s3.progress.TransferListener;

import static org.assertj.core.api.Assertions.assertThat;

public class S3TransferFileWithProgressTests {

	static final LocalStackContainer LOCAL_STACK_CONTAINER =
			new LocalStackContainer(
					DockerImageName.parse("localstack/localstack:1.4.0"))
					.withServices(LocalStackContainer.Service.S3);

	@TempDir
	static Path temporaryFolder;

	@BeforeAll
	static void startContainer() {
		LOCAL_STACK_CONTAINER.start();
	}

	@Test
	void theTransferListenerIsUpdatedWhenFileHasBeenUploaded() throws IOException, InterruptedException {
		S3AsyncClient s3AsyncClient =
				S3AsyncClient.builder()
						.region(Region.of(LOCAL_STACK_CONTAINER.getRegion()))
						.credentialsProvider(StaticCredentialsProvider.create(
								AwsBasicCredentials.create(LOCAL_STACK_CONTAINER.getAccessKey(),
										LOCAL_STACK_CONTAINER.getSecretKey())))
						.endpointOverride(LOCAL_STACK_CONTAINER.getEndpointOverride(LocalStackContainer.Service.S3))
						.build();

		S3TransferManager s3TransferManager = S3TransferManager.builder().s3Client(s3AsyncClient).build();

		File file = new File(temporaryFolder.toFile(), "test.txt");
		file.createNewFile();
		byte[] testData = "test data".getBytes();
		try (OutputStream out = new FileOutputStream(file)) {
			out.write(testData);
		}

		CountDownLatch transferInitiated = new CountDownLatch(1);
		CountDownLatch bytesTransferred = new CountDownLatch(1);
		CountDownLatch transferComplete = new CountDownLatch(1);

		TransferListener transferListener = new TransferListener() {
			@Override
			public void transferInitiated(Context.TransferInitiated context) {
				transferInitiated.countDown();
			}

			@Override
			public void bytesTransferred(Context.BytesTransferred context) {
				bytesTransferred.countDown();
			}

			@Override
			public void transferComplete(Context.TransferComplete context) {
				transferComplete.countDown();
			}

		};

		String testBucket = "test-bucket";

		s3AsyncClient.createBucket(request -> request.bucket(testBucket)).join();

		UploadRequest.Builder uploadRequest =
				UploadRequest.builder()
						.putObjectRequest(request -> request.bucket(testBucket).key(file.getName()))
						.requestBody(AsyncRequestBody.fromFile(file))
//						.requestBody(AsyncRequestBody.fromBytes(testData))
						.addTransferListener(transferListener)
						.addTransferListener(LoggingTransferListener.create());

		s3TransferManager.upload(uploadRequest.build()).completionFuture().join();

		assertThat(transferInitiated.await(10, TimeUnit.SECONDS)).isTrue();
		assertThat(bytesTransferred.await(10, TimeUnit.SECONDS)).isTrue();
		assertThat(transferComplete.await(10, TimeUnit.SECONDS)).isTrue();

		ResponseBytes<GetObjectResponse> responseBytes =
				s3AsyncClient.getObject(request -> request.bucket(testBucket).key(file.getName()),
								AsyncResponseTransformer.toBytes())
						.join();

		assertThat(responseBytes.asByteArray()).isEqualTo(testData);
	}

}

Uncomment AsyncRequestBody.fromBytes() line and you'll see the test passing.

Possible Solution

No response

Additional Information/Context

No response

AWS Java SDK version used

2.20.22

JDK version used

openjdk version "17.0.5" 2022-10-18

Operating System and version

Windows 10

[debora-ito]

@artembilan thank you for reporting this, bug acknowledged.

[zoewangg]

This is fixed in #3839, could you try with the latest version?

[artembilan]

Confirmed as working now.

You can close issue respectively.

Thanks

[github-actions]

This issue is now closed. Comments on closed issues are hard for our team to see.
If you need more assistance, please open a new issue that references this one.