Flush buffered acknowledgements below threshold on shutdown

[hyeongguen-song]

📢 Type of change

• Bugfix
• New feature
• Enhancement
• Refactoring

📜 Description

Fixes #1661.

BatchingAcknowledgementProcessor can drop buffered acknowledgements on
shutdown. Whatever is left in the buffer below acknowledgementThreshold is
never flushed, so the processor spins for the whole
acknowledgementShutdownTimeout, logs

Acknowledgements did not finish in 20000 ms. Proceeding with shutdown.

and then clears the buffer. Those messages are never deleted from SQS and are
redelivered once the visibility timeout expires.

The scheduling condition is aligned with shouldKeepPollingAcks():

private boolean shouldKeepSchedulingAcks() {
    return this.context.isRunning() || !this.context.isTimeoutElapsed;
}

💡 Motivation and Context

AbstractOrderingAcknowledgementProcessor.stop() sets running = false before
calling doStop(). From that moment scheduleNextExecution() refuses to arm the
next execution:

private void scheduleNextExecution(Instant nextExecutionDelay) {
    if (!this.context.isRunning()) {   // <-- here
        return;

The polling thread, however, keeps running until the shutdown timeout elapses
(shouldKeepPollingAcks()), so it goes on moving messages from the ack queue
into the buffer, and each threshold execution pushes lastAcknowledgement
forward via doExecute().

That leaves exactly one already-armed scheduled execution, whose fire time is
T_arm + ackInterval. Because every threshold execution necessarily happens
after T_arm, lastAcknowledgement is always newer than T_arm, so the guard
in pollAndExecuteScheduled()

if (Instant.now().isAfter(this.context.lastAcknowledgement.plus(this.ackInterval))) {

is deterministically false. That execution skips the flush and cannot re-arm, so
nothing is left to empty the buffer -- the threshold path would need a full
acknowledgementThreshold of new messages, and polling has stopped supplying
them.

The two conditions are meant to cover the same drain window; only one of them was
updated when #1082 added the shutdown wait. The task scheduler is still alive
during that window, since doStop() only disposes of it after
waitAcknowledgementsToFinish() returns.

This is load-dependent, which is why it is easy to miss. It needs the threshold
path to be the dominant flush path, i.e. messages arriving faster than one
acknowledgementInterval per batch. We hit it in production after a scaling
change raised per-pod throughput several-fold: pod terminations started leaving a
small number of messages undeleted, visible as a gap between
NumberOfMessagesReceived and NumberOfMessagesDeleted and as a spike in
ApproximateAgeOfOldestMessage matching the visibility timeout.

💚 How did you test it?

Added givenScheduledAcknowledgement_whenStoppedWithMessagesBelowThreshold_shouldAcknowledgeRemainingMessages.

The three existing shutdown tests cannot catch this: they all use
acknowledgementInterval(Duration.ZERO), and ScheduledAcknowledgementExecution.start()
only arms anything when ackInterval != Duration.ZERO, so the scheduled path is
never active. They also use exactly 100 messages against a threshold of 10, so
no remainder is ever left in the buffer.

The new test uses a non-zero interval and 105 messages against a threshold of 10.
Without the fix:

Tests run: 1, Failures: 1 -- Time elapsed: 10.86 s
Expecting actual: [...]
but could not find the following elements: [payload=100 .. payload=104]

The 10.86 s is the full acknowledgementShutdownTimeout being spun, and the five
missing messages are exactly the sub-threshold remainder.

With the fix, ./mvnw -pl spring-cloud-aws-sqs -am test (JDK 17.0.19) gives
635 tests, 0 failures, 0 errors, 6 skipped, spotless checks enabled,
including the LocalStack integration tests.

Note that CI on this branch will fail to compile until #1662 is merged --
main currently does not build. I verified the combined state (this branch plus
#1662) locally, which is what CI will see once that lands; happy to rebase then.

📝 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

Users who cannot wait for a release can set acknowledgementInterval(Duration.ZERO)
to get an ImmediateAcknowledgementProcessor instead, which has no buffer and no
scheduler. With the standard-SQS defaults (acknowledgementInterval 1s,
acknowledgementThreshold 10) the exposure window is up to
acknowledgementThreshold - 1 messages per shutdown, per pod.

[tomazfernandes]

Thanks @hyeongguen-song, great analysis and writeup. I have three asks so we can close this class of bug for good if you're up for it.

1. First is for us to flush from the shutdown wait instead of keeping the scheduler armed. We'd drop the shouldKeepSchedulingAcks() change, and instead, in waitOnAcknowledgementsIfTimeoutSet()'s loop: if acks.isEmpty(), take the lock and executeAllAcks() to flush any remaining acks in the buffer,

This would make the drain a guarantee of the shutdown path itself, and also covers interval = ZERO + threshold > 0, where no scheduled execution ever exists and a remainder is lost on every shutdown.

This would need to be inside the loop: the polling thread can hold a just-polled message when the queue looks empty.

What do you think?

2. Let's make running volatile in AbstractOrderingAcknowledgementProcessor. It's read unsynchronized from the polling and scheduler threads. The current test you added is nondeterministic: with the fix reverted it fails only ~1 in 3 runs here because of stale true reads.

3. Let's add these test scenarios:

• Test A: sub-threshold remainder with a live timer config:

1. Ack exactly threshold (10) messages.
2. Await that threshold flush completing (latch in the acknowledgement executor).
3. Ack a sub-threshold remainder (e.g. 5).
4. stop().
5. Assert all 15 messages were acked

• B: interval = ZERO, threshold = 10, ack 15, stop(), assert all 15 acked.

Let me know your thoughts.