Fix residual leak in downloadDirectory by skipping file downloads after cancellation.

[RanVaknin]

In #6875, we addressed canceling in flight transfers when a directory download is cancelled. That fix reduced the leaked files per orphaned directory from thousands to single digits in stress test. However there was a residual leak for downloads that were already dispatched to doDownloadSingleFile (the method that creates the destination directory and initiates the file download) before the cancellation signal propagated.

When cancellation occurs, thread A calls subscription.cancel() to stop new items from being delivered. Meanwhile, thread B has already picked up an item from onNext and is inside doDownloadSingleFile. Since it entered before thread A's cancellation took effect, it creates the destination directory and starts a download into it.

This PR adds a isDone() guard at the top of doDownloadSingleFile so that any thread that entered the method before cancellation took effect will return early before touching the filesystem.

The two tests:

1. Updated downloadDirectory_cancel_shouldCancelAllFutures - now waits for both downloads to start before cancelling. The original test called cancel() immediately after downloadDirectory(), implicitly assuming all downloads would always start regardless of cancellation state. With the new isDone() guard, that assumption is not always true since cancel can now prevent downloads from starting. The test's intent is unchanged (in flight futures get cancelled), we just make sure both futures are actually in flight before cancelling. (Sanity check tested this with a repeated test suite 1000 times to make sure its deterministic)

2. Added downloadDirectory_cancelledFuture_shouldNotCreateDirectories - The test cancels the directory download future, then delivers an S3 object through the publisher. Asserts that the destination subdirectory was never created on the filesystem. Without the fix, doDownloadSingleFile would call createParentDirectoriesIfNeeded() and create the directory despite the cancellation.

[zoewangg]

It seems the race will still be there. Curious why existing request cancellation logic doesn't work. https://github.com/aws/aws-sdk-java-v2/blob/master/services-custom/s3-transfer-manager/src/main/java/software/amazon/awssdk/transfer/s3/internal/AsyncBufferingSubscriber.java#L56-L63

[RanVaknin]

Curious why existing request cancellation logic doesn't work.

I also was wondering about it. I saw directories being recreated and populated with new files after the cancellation hook fired. The only explanation I could think of is that the race is happening between a thread that has that cancellation signal and one that doesnt.

1. If a new thread B enters onNext() and hit consumer.apply(item)
2. Thread A sends the cancellation signal requestsInFlight.forEach(f -> f.cancel(true))
3. Thread B hits requestsInFlight.add(currentRequest)

In this case thread A cancellation wont stop thread B because it didnt have a time to register the request future with the requestsInFlight set.

It seems the race will still be there

what do you mean?