fix(sqs): stop close() busy-loop duplicate sends

[zoewangg]

Motivation and Context

This fixes two shutdown issues in SqsAsyncBatchManager.close(), both in the shared internal base class RequestBatchManager (module services/sqs).

1. Redundant re-sends of a buffered batch on close. When close() was called while other threads were still submitting on a real asynchronous client, a buffered batch could be sent more than once: close() waited for the in-flight response by re-sending the same batch, because the drain loop stopped only after an asynchronous completion callback cleared it rather than after draining the buffer. The effect was a bounded number of duplicate SendMessageBatch calls for the same messages during that single close(). close() now drains each buffered batch and sends it exactly once, stopping when the buffer is empty (partial batches included).

2. In-flight requests cancelled on shutdown. Previously close() cancelled in-flight batch sends immediately, so a request that had already been dispatched could complete with CancellationException instead of its real SendMessageResponse. close() now waits a bounded grace period for in-flight sends to finish so their callers receive the real result, and cancels only the requests still outstanding when that period expires. This mirrors the graceful bounded-await shutdown already used by CloudWatchMetricPublisher.close().

Modifications

Drain each buffered batch exactly once:

• RequestBatchManager now drains each buffer by re-extracting the next batch with extractEntriesForScheduledFlush (the drain-any-non-empty primitive performScheduledFlush already uses) and flushing each distinct batch exactly once, terminating when the buffer is empty. Full and partial batches are now sent exactly once (previously a residual partial batch was never sent and its callers were cancelled).

Graceful bounded-await shutdown (dispatch-all-first, then a single shared wait):

• RequestBatchManager exposes a two-phase internal shutdown surface: dispatchPending() (phase 1, non-blocking) marks the manager closed, drains/dispatches every buffered batch, and returns a CompletableFuture<Void> that completes when the in-flight sends it dispatched complete; cancelPending() (phase 2) cancels anything still pending and releases the buffers. The wait targets the caller-facing futures (pendingResponses), so a completing send delivers the real result rather than being cancelled.
• DefaultSqsAsyncBatchManager.close() orchestrates the shutdown: it calls dispatchPending() on all three write managers first (every buffered batch goes on the wire at once), then performs a single bounded wait via CompletableFuture.allOf(...) up to one shared grace period, then calls cancelPending() on all three, then closes the receive manager. Because all three dispatch before the single wait, the whole shutdown is bounded by ONE grace period, not one per manager. The wait helper never throws (handles TimeoutException/ExecutionException/InterruptedException, restoring the interrupt flag).
• The grace period is a new internal field on RequestBatchConfiguration (@SdkInternalApi), default 5 seconds. There is no public knob.
• DefaultSqsAsyncBatchManager.close() is guarded by an AtomicBoolean so it is idempotent and safe under concurrent calls: only the first caller runs the dispatch/await/cancel orchestration, and a concurrent or second call is a clean no-op (this avoids a concurrent loser cancelling the winner's still-awaited in-flight sends).
• RequestBatchManager gains an AtomicBoolean closed guard: a sendMessage / deleteMessage / changeMessageVisibility submitted after close returns a future completed exceptionally with IllegalStateException, matching the receive path.

Documentation:

• Added an @Override void close() Javadoc on the public SqsAsyncBatchManager documenting that close flushes buffered requests, blocks up to a bounded grace period for in-flight sends so callers receive real results, then cancels stragglers.

Testing

services/sqs built with full static analysis (Checkstyle, SpotBugs, PMD, japicmp) - BUILD SUCCESS, no public API break. New and existing tests pass.

Added new test cases

• DefaultSqsAsyncBatchManagerTest - close/shutdown behavior against the REAL DefaultSqsAsyncBatchManager (and, through it, the real write batch managers) over a mock SqsAsyncClient with controllable batch-send futures:
• SqsAsyncBatchManagerTest (WireMock) - real end-to-end batching over HTTP.

Types of changes

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

Checklist

• I have read the CONTRIBUTING document
• Local run of mvn install succeeds (built the affected module services/sqs, not a full-repo mvn install)
• My code follows the code style of this project
• My change requires a change to the Javadoc documentation
• I have updated the Javadoc documentation accordingly
• I have added tests to cover my changes
• All new and existing tests passed
• I have added a changelog entry.
• My change is to implement 1.11 parity feature and I have updated LaunchChangelog

License

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