fix: netty mem leak during failure

[ShubhamChaturvedi7]

Issue #, if available:

Description of changes:
Drain the stream if a failure has occurred after fetching the payload.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Check any applicable:

• Were any files moved? Moving files changes their URL, which breaks all hyperlinks to the files.

[josecorella]

since we are no longer throwing this exception, how are we notifying to custoer that the decryption materials are null and that they may have a misconfigured cmm? removing this exception can be a breaking change, no?

[ShubhamChaturvedi7]

The comments explain that - since this is async pipeline, a prev step failed and the Future has exception set. Throwing from here doesn't throw it for the callers, but rather for the pipeline. Which is exactly what was causing the memory leak.

TLDR: The caller still sees the exception - see the test case here

[lucasmcdonald3]

Suggested change

	// prepare() runs once per request attempt (AsyncResponseTransformer#prepare, enforced by
	// BaseAsyncClientHandler via IdempotentAsyncResponseHandler keyed on EXECUTION_ATTEMPT).
	// Clearing materials is what makes onStream below take the drain path when this attempt's
	// onResponse fails, instead of decrypting this attempt's body with the previous attempt's
	// materials. onResponse re-resolves materials unconditionally, so this costs no extra CMM call.
	materials = null;
	shouldDrain = true;

[ShubhamChaturvedi7]

This won't work. We don't need to drain on every call. We only need to drain if materials fail to fetch.

For every prepare call, the SDK also calls onResponse, which either overrides materials or throws. If it overrides, the flow executes normally. If it throws, materials is null then only we should drain.

Essentially we would need to atomically switch shouldDrain based on whether materials is null, which adds unnecessary complication.

A compromise would be to not set materials to null here (like how it was before this change). The flow would then either override it in onResponse or throw. That can cause memory leak in the scenario where the material fetch succeeded on the first attempt but then an exception occurred, SDK retried it but again an exception occurred (so now materials in not null from first attempt, and hence our drain won't fire).

[lucasmcdonald3]

Suggested change

	var shouldDrain = false;

[lucasmcdonald3]

Suggested change

	public void onStream(SdkPublisher<ByteBuffer> ciphertextPublisher) {
	if (materials == null) {
	public void onStream(SdkPublisher<ByteBuffer> ciphertextPublisher) {
	if (shouldDrain) { /* drain, etc. */ }
	if (materials == null) { /* throw existing exception */ }

[ShubhamChaturvedi7]

We can never throw from here. See my explanation above. Any error is handled in the exceptionOccurred block so we can only throw once which is done in the prepareMaterialsFromRequest block. Throwing again from here goes into the nettry pipeline.

[lucasmcdonald3]

This PR uses "is materials empty" to mean "did an attempt fail", but that 1) is more clearly expressed as a separate check of "an attempt failed, drain the body", 2) collapses into an existing case