Fix potentially flaky S3TransferManagerListenerTests (#2801)

* Fix potentially flaky S3TransferManagerListenerTests

As they were previously written, these tests were subject to fail due to
the following reasons:

1. The execution order of multiple `CompletableFuture#whenComplete(..)`
calls is non-deterministic (unless the result of each one is chained
together, in sequence)
2. Therefore it was possible that the `CompletableFuture` associated
with a transfer may be completed before or after
`TransferListener#transferComplete(..)` was invoked
3. While this is not problematic from a public API perspective, our
tests assumed that the future completion meant all listeners had been
exercised, which may not always be true

This change fixes the tests to instead await for the expected
interactions to occur, rather than await for the associated future to be
completed.

An alternative approach would be to sequence the multiple completion
stages together, guaranteeing that listeners are invoked before the
future is completed, but this may delay the completion of the future in
the event of longer-running listener implementations. It seems
preferable to favor immediate completion of the future.

Author: Bennett-Lynch

.changes/next-release/bugfix-S3TransferManager-e4998ee.json

@@ -0,0 +1,6 @@
+{
+    "category": "S3TransferManager", 
+    "contributor": "", 
+    "type": "bugfix", 
+    "description": "Fix potentially flaky S3TransferManagerListenerTests"
+}

services-custom/s3-transfer-manager/src/test/java/software/amazon/awssdk/transfer/s3/internal/S3TransferManagerListenerTest.java

@@ -20,7 +20,7 @@
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -93,33 +93,33 @@ public void upload_success_shouldInvokeListener() throws Exception {
                                                    .overrideConfiguration(b -> b.addListener(listener))
                                                    .build();
         Upload upload = tm.upload(uploadRequest);
-        upload.completionFuture().join();
 
         ArgumentCaptor<TransferListener.Context.TransferInitiated> captor1 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferInitiated.class);
-        verify(listener, times(1)).transferInitiated(captor1.capture());
+        verify(listener, timeout(1000).times(1)).transferInitiated(captor1.capture());
         TransferListener.Context.TransferInitiated ctx1 = captor1.getValue();
         assertThat(ctx1.request()).isSameAs(uploadRequest);
         assertThat(ctx1.progressSnapshot().transferSizeInBytes()).hasValue(contentLength);
         assertThat(ctx1.progressSnapshot().bytesTransferred()).isZero();
 
         ArgumentCaptor<TransferListener.Context.BytesTransferred> captor2 =
             ArgumentCaptor.forClass(TransferListener.Context.BytesTransferred.class);
-        verify(listener, times(1)).bytesTransferred(captor2.capture());
+        verify(listener, timeout(1000).times(1)).bytesTransferred(captor2.capture());
         TransferListener.Context.BytesTransferred ctx2 = captor2.getValue();
         assertThat(ctx2.request()).isSameAs(uploadRequest);
         assertThat(ctx2.progressSnapshot().transferSizeInBytes()).hasValue(contentLength);
         assertThat(ctx2.progressSnapshot().bytesTransferred()).isPositive();
 
         ArgumentCaptor<TransferListener.Context.TransferComplete> captor3 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferComplete.class);
-        verify(listener, times(1)).transferComplete(captor3.capture());
+        verify(listener, timeout(1000).times(1)).transferComplete(captor3.capture());
         TransferListener.Context.TransferComplete ctx3 = captor3.getValue();
         assertThat(ctx3.request()).isSameAs(uploadRequest);
         assertThat(ctx3.progressSnapshot().transferSizeInBytes()).hasValue(contentLength);
         assertThat(ctx3.progressSnapshot().bytesTransferred()).isEqualTo(contentLength);
         assertThat(ctx3.completedTransfer()).isSameAs(upload.completionFuture().get());
 
+        upload.completionFuture().join();
         verifyNoMoreInteractions(listener);
     }
 
@@ -134,11 +134,10 @@ public void download_success_shouldInvokeListener() throws Exception {
                                                          .overrideConfiguration(b -> b.addListener(listener))
                                                          .build();
         Download download = tm.download(downloadRequest);
-        download.completionFuture().join();
 
         ArgumentCaptor<TransferListener.Context.TransferInitiated> captor1 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferInitiated.class);
-        verify(listener, times(1)).transferInitiated(captor1.capture());
+        verify(listener, timeout(1000).times(1)).transferInitiated(captor1.capture());
         TransferListener.Context.TransferInitiated ctx1 = captor1.getValue();
         assertThat(ctx1.request()).isSameAs(downloadRequest);
         // transferSize is not known until we receive GetObjectResponse header
@@ -147,7 +146,7 @@ public void download_success_shouldInvokeListener() throws Exception {
 
         ArgumentCaptor<TransferListener.Context.BytesTransferred> captor2 =
             ArgumentCaptor.forClass(TransferListener.Context.BytesTransferred.class);
-        verify(listener, times(1)).bytesTransferred(captor2.capture());
+        verify(listener, timeout(1000).times(1)).bytesTransferred(captor2.capture());
         TransferListener.Context.BytesTransferred ctx2 = captor2.getValue();
         assertThat(ctx2.request()).isSameAs(downloadRequest);
         // transferSize should now be known
@@ -156,13 +155,14 @@ public void download_success_shouldInvokeListener() throws Exception {
 
         ArgumentCaptor<TransferListener.Context.TransferComplete> captor3 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferComplete.class);
-        verify(listener, times(1)).transferComplete(captor3.capture());
+        verify(listener, timeout(1000).times(1)).transferComplete(captor3.capture());
         TransferListener.Context.TransferComplete ctx3 = captor3.getValue();
         assertThat(ctx3.request()).isSameAs(downloadRequest);
         assertThat(ctx3.progressSnapshot().transferSizeInBytes()).hasValue(contentLength);
         assertThat(ctx3.progressSnapshot().bytesTransferred()).isEqualTo(contentLength);
         assertThat(ctx3.completedTransfer()).isSameAs(download.completionFuture().get());
 
+        download.completionFuture().join();
         verifyNoMoreInteractions(listener);
     }
 
@@ -188,7 +188,7 @@ public void upload_failure_shouldInvokeListener() throws Exception {
 
         ArgumentCaptor<TransferListener.Context.TransferInitiated> captor1 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferInitiated.class);
-        verify(listener, times(1)).transferInitiated(captor1.capture());
+        verify(listener, timeout(1000).times(1)).transferInitiated(captor1.capture());
         TransferListener.Context.TransferInitiated ctx1 = captor1.getValue();
         assertThat(ctx1.request()).isSameAs(uploadRequest);
         // transferSize is not known since file did not exist
@@ -197,13 +197,14 @@ public void upload_failure_shouldInvokeListener() throws Exception {
 
         ArgumentCaptor<TransferListener.Context.TransferFailed> captor2 =
             ArgumentCaptor.forClass(TransferListener.Context.TransferFailed.class);
-        verify(listener, times(1)).transferFailed(captor2.capture());
+        verify(listener, timeout(1000).times(1)).transferFailed(captor2.capture());
         TransferListener.Context.TransferFailed ctx2 = captor2.getValue();
         assertThat(ctx2.request()).isSameAs(uploadRequest);
         assertThat(ctx2.progressSnapshot().transferSizeInBytes()).isNotPresent();
         assertThat(ctx2.progressSnapshot().bytesTransferred()).isZero();
         assertThat(ctx2.exception()).isInstanceOf(NoSuchFileException.class);
 
+        upload.completionFuture().join();
         verifyNoMoreInteractions(listener);
     }
 
@@ -221,11 +222,12 @@ public void listener_exception_shouldBeSuppressed() throws Exception {
                                                    .overrideConfiguration(b -> b.addListener(listener))
                                                    .build();
         Upload upload = tm.upload(uploadRequest);
-        upload.completionFuture().join();
 
-        verify(listener, times(1)).transferInitiated(any());
-        verify(listener, times(1)).bytesTransferred(any());
-        verify(listener, times(1)).transferComplete(any());
+        verify(listener, timeout(1000).times(1)).transferInitiated(any());
+        verify(listener, timeout(1000).times(1)).bytesTransferred(any());
+        verify(listener, timeout(1000).times(1)).transferComplete(any());
+
+        upload.completionFuture().join();
         verifyNoMoreInteractions(listener);
     }