Account for acknowledgements not yet added to the buffer on shutdown

[hyeongguen-song]

📢 Type of change

• Bugfix
• New feature
• Enhancement
• Refactoring

📜 Description

Follow-up to #1663, for the window we discussed there:

Yes, let's squash this one as well. It's a narrow window that can only affect one
message per shutdown, but as you said under contention it widens up.

hasAcksLeft() decided whether the shutdown wait was done by looking at the ack
queue and the buffer. A message the polling thread has already taken off the queue
but not yet added to the buffer is in neither of them, so the wait could end while
such a message was in flight.

💡 Motivation and Context

Message<T> polledMessage = this.acks.poll(1, TimeUnit.SECONDS);   // removed from acks
if (polledMessage != null) {
    addMessageToBuffer(polledMessage);                            // not in the buffer yet

If the wait loop sampled in between, it saw nothing left and returned.
waitAcknowledgementsToFinish() then set isTimeoutElapsed and cleared the
buffer, and the polling thread added the message to a buffer nothing would ever
flush.

Unlike the case fixed in #1663, this one produces no warning at all — the wait
loop exits normally rather than timing out, so there is nothing in the logs to
suggest a message was dropped. In the test below the unfixed run finishes in 1.6 s
instead of spinning the 10 s shutdown timeout.

At most one message per shutdown can be in this window, since the polling thread
handles one at a time. It widens whenever addMessageToBuffer() has to wait on the
buffer lock, which happens while an execution is being dispatched — and for
AcknowledgementOrdering.ORDERED that dispatch takes the ordered execution lock
while holding the buffer lock.

The fix tracks the messages that have been received but are in neither the queue
nor the buffer, and includes them in hasAcksLeft(). The counter is incremented
before the message is offered to the queue and decremented after it has been added
to the buffer, so it is never undercounted. It is briefly double counted while the
message sits in the buffer, which only makes the wait more conservative.

hasAcksLeft() now uses the counter in place of acks.size(), since the counter
is always greater than or equal to it.

I left flushRemainingAcks() gated on acks.isEmpty() as it is. Its purpose is to
avoid splitting batches, not to decide when the wait is over, and if it does flush
while a message is in flight that message is simply picked up by the next iteration
now that hasAcksLeft() accounts for it.

💚 How did you test it?

givenMessagePolledButNotBufferedYet_whenStopped_shouldAcknowledgeIt reproduces
the window deterministically through a public extension point rather than by
reaching into internals. The message grouping function is applied inside
addMessageToBuffer() while the buffer lock is held, so a grouping function that
blocks parks the polling thread at exactly the point where the message is in
neither the queue nor the buffer. The test then calls stop() from another thread,
gives the wait loop several 200 ms iterations to sample, and releases the polling
thread.

Measured on JDK 17.0.19, 15 consecutive runs each:

	result
with this change	15/15 pass
main	15/15 reproduce the bug

On main the assertion shows no acknowledgement was executed at all:

Expecting actual:
  []
to contain exactly (and in same order):
  [GenericMessage [payload=0, ...]]

Full module: ./mvnw -pl spring-cloud-aws-sqs -am test gives 637 tests, 0
failures, 0 errors, 6 skipped, spotless checks enabled, including the LocalStack
integration tests.

📝 Checklist

• I reviewed submitted code
• I added tests to verify changes
• I updated reference documentation to reflect the change
• All tests passing
• No breaking changes

No documentation change needed: this restores the documented behaviour rather than
changing it.

🔮 Next steps

Nothing further from me on this path. The remaining structural asymmetry is that
the polling thread's lifetime is bounded by isTimeoutElapsed, which is only set
after the wait loop has already decided it is finished — so correctness relies on
the wait loop accounting for everything the polling thread might still be holding,
rather than on the polling thread having demonstrably stopped. Making the wait join
the polling thread would remove that reliance, but it is a bigger change than this
one and I did not want to fold it in.

[hyeongguen-song]

@tomazfernandes this is the follow-up you asked for in #1663. Short version:

hasAcksLeft() checked acks.size() and acksBuffer.size(). A message the
polling thread has polled but not yet added to the buffer is in neither, so the
shutdown wait could return while one was in flight, the buffer was then cleared,
and the message was added to it afterwards and never acknowledged.

The fix counts messages that have been received but are in neither the queue nor
the buffer, and includes that in hasAcksLeft(). Incremented before the offer,
decremented after the buffer add, so it is never undercounted — the brief double
counting while the message sits in the buffer only makes the wait more
conservative.

Two things worth your attention:

• Unlike the case in Flush buffered acknowledgements from the shutdown wait #1663, this one leaves nothing in the logs. The wait loop
  exits normally instead of timing out, so there is no warning at all. In the test
  the unfixed run finishes in 1.6 s rather than spinning the 10 s timeout, with
  zero acknowledgements executed.
• I left flushRemainingAcks() gated on acks.isEmpty() rather than on the new
  counter. Its job is to avoid splitting batches, not to decide when the wait is
  over, and a message that arrives after a flush is now picked up by the next
  iteration. Happy to change it if you'd rather have both use the counter.

The test drives the window through setMessageGroupingFunction, which is applied
inside addMessageToBuffer() while the buffer lock is held, so blocking in it
parks the polling thread exactly where the message is in neither place. No
internals are touched. 15/15 pass with the change, 15/15 reproduce on main, and
the full module is green at 637 tests.

One structural note I put in the description rather than acting on: the polling
thread's lifetime is bounded by isTimeoutElapsed, which is only set after the
wait loop has already decided it is done. So correctness still rests on the wait
loop accounting for everything the polling thread might be holding, rather than on
the polling thread having demonstrably stopped. Having the wait join the polling
thread would remove that, but it is a much bigger change and I did not want to
fold it in here.

[tomazfernandes]

Thanks for the PR @hyeongguen-song. After this PR, shutdown behavior is correct for all known windows.

I agree the join design would make correctness structural rather than bookkeeping every transition. Since behavior is stable now, let's keep it in mind for the next major version.