Fix intermittent NPE race in EmittingSubscription when cancel() nulls subscriber during emit

[joviegas]

Fixes #7351

Issue

A customer downloading a small object with a multipart-enabled async S3 client (getObject/downloadFile with multipartEnabled(true)) intermittently gets a NullPointerException instead of their object. The same call succeeds on retry, and the object is small enough to need no multipart at all. It shows up more on fast endpoints and busy hosts, so it hits local testing and CI hardest.

Root cause

EmittingSubscription.downstreamSubscriber is a non-volatile field shared by two threads. For a single-part object, the caller thread runs the emit loop while a Netty thread, seeing partsCount == null, calls cancel(), which sets the field to null. The loop checks isCancelled before onNext, but the cancel can land in the gap between that check and the dereference, so onNext is called on a null reference and throws.

Fix

EmittingSubscription: make downstreamSubscriber final and stop nulling it in cancel(). Cancellation is already signaled by the existing isCancelled flag, so the reference no longer needs to be cleared. A final reference cannot be observed as null.

FileAsyncResponseTransformerPublisher (its only user, same defect one frame away): this class is the Publisher, so Reactive Streams rule 3.13 requires it to drop the subscriber on cancel. It keeps nulling, but the field is now volatile and every read snapshots it into a local and null-checks before use.

Testing

• New EmittingSubscriptionTest: demand/emit, non-positive demand, cancel, and a concurrency test racing cancel() against the emit loop. The concurrency test fails on the old code and passes on the fix.
• Existing FileAsyncResponseTransformerPublisher unit tests and its Reactive Streams TCK suite pass, including the rule 3.13 test that verifies the publisher drops the subscriber on cancel (the behavior the publisher change touches).

Screenshots (if appropriate)

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)

License

• I confirm that this pull request can be released under the Apache 2 license