Flush buffered acknowledgements from the shutdown wait

[hyeongguen-song]

📢 Type of change

• Bugfix
• New feature
• Enhancement
• Refactoring

📜 Description

Fixes #1661.

Messages left in the acknowledgement buffer below acknowledgementThreshold are
never flushed on shutdown. The processor spins for the whole
acknowledgementShutdownTimeout, logs

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

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

The buffer is now drained from the shutdown wait loop:

private void flushRemainingAcks() {
    if (!this.acks.isEmpty()) {
        return;
    }
    this.context.lock();
    try {
        this.context.executeAllAcks();
    }
    finally {
        this.context.unlock();
    }
}

running in AbstractOrderingAcknowledgementProcessor is also made volatile.

💡 Motivation and Context

There are two independent ways to lose the remainder, and draining from the
shutdown wait covers both:

With a scheduled execution. AbstractOrderingAcknowledgementProcessor.stop()
sets running = false before calling doStop(), so scheduleNextExecution()
refuses to arm the next execution. The polling thread keeps running until the
shutdown timeout elapses (shouldKeepPollingAcks()), so it goes on moving
messages into the buffer, and each threshold execution pushes
lastAcknowledgement forward. The one already-armed execution then fires with
lastAcknowledgement newer than the time it was armed, skips the flush because
now is not after lastAcknowledgement + ackInterval, and cannot re-arm.

With no scheduled execution at all. ScheduledAcknowledgementExecution.start()
only arms anything when ackInterval != Duration.ZERO. With
acknowledgementInterval = ZERO and a positive threshold — a configuration
doStart() explicitly allows — the threshold path is the only flush mechanism,
so a sub-threshold remainder is lost on every shutdown.

Making the drain part of the shutdown path itself covers both, instead of
depending on a scheduled execution existing and being armed.

Two details in the implementation:

• The flush waits for the ack queue to drain, so batches are not needlessly split.
• It is retried on every iteration of the wait loop rather than executed once. The
  polling thread may be holding a message it has already polled but not yet added
  to the buffer, in which case the queue looks empty while the message is in
  neither the queue nor the buffer.

running was written under lifecycleMonitor but read unsynchronized from the
polling and scheduler threads via isRunning(), so neither thread had any
guarantee of observing stop().

This is load-dependent in the scheduled case, 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?

Two new tests, one per loss path. Both are behaviour-only and do not reach into
internals:

• givenScheduledAcknowledgement_whenStoppedWithRemainderBelowThreshold_... acks
  exactly acknowledgementThreshold messages, waits on a latch in the
  acknowledgement executor so the threshold flush is known to have completed, then
  acks a sub-threshold remainder and stops. The interval outlasts the shutdown
  timeout, so a scheduled execution exists but provably cannot fire during the
  test.
• givenZeroIntervalAndThreshold_whenStoppedWithRemainderBelowThreshold_... uses
  acknowledgementInterval = ZERO, where no scheduled execution exists.

Both fail on main and pass with this change. Measured on JDK 17.0.19, 15
consecutive runs each:

	result
with this change	15/15 pass
main + volatile only	15/15 reproduce the bug
main unchanged	15/15 reproduce the bug

On failure each test spins the full 10 s shutdown timeout and loses exactly the
5-message remainder.

The three existing shutdown tests could not catch either path: they all use
acknowledgementInterval(Duration.ZERO) so no scheduled execution is ever
created, and they use exactly 100 messages against a threshold of 10, so no
remainder is ever left in the buffer.

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

Rebased onto main after #1662 landed, and re-verified with no local patches
applied.

📝 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 switch to an
ImmediateAcknowledgementProcessor, which has no buffer and no scheduler, by
setting acknowledgementInterval(Duration.ZERO) and leaving
acknowledgementThreshold unset — StandardSqsComponentFactory selects it when
the interval is ZERO and the threshold is null or 0. Anyone who has
explicitly configured a positive threshold needs to set it back to 0 as well,
otherwise they still get the batching processor and the second path above.

[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.

[hyeongguen-song]

@tomazfernandes

All three make sense to me — pushed. Details below.

1. Flush from the shutdown wait. Agreed, and your interval = ZERO point is
the decisive one: my approach did nothing for that configuration. I wrote your
Test B against my previous commit to check, and it failed exactly as you'd
expect — full 10 s spin, remainder lost:

givenZeroIntervalAndThreshold_... FAILED  (Time elapsed: 10.82 s)
could not find: payload=10, 11, 12, 13, 14

ScheduledAcknowledgementExecution.start() only arms when
ackInterval != Duration.ZERO, so there was nothing for shouldKeepSchedulingAcks()
to keep alive. Making the drain part of the shutdown path covers both, so I
dropped that change entirely.

Implemented as you described, gated on acks.isEmpty() and inside the loop:

private void flushRemainingAcks() {
    if (!this.acks.isEmpty()) {
        return;
    }
    this.context.lock();
    try {
        this.context.executeAllAcks();
    }
    finally {
        this.context.unlock();
    }
}

The futures go through manageFuture(), so hasUnfinishedAcks() sees them and
the wait loop keeps waiting until the flush actually completes.

2. volatile. Done, with a comment on the field. Agreed on the reasoning —
the write is under lifecycleMonitor but isRunning() reads it unsynchronized,
so neither the polling nor the scheduler thread had a guarantee of observing
stop().

I could not reproduce the ~1-in-3 rate, though. On JDK 17.0.19 / arm64, 15
consecutive runs each:

	result
with this change	15/15 pass
main + volatile only	15/15 reproduce
main unchanged	15/15 reproduce

So the visibility of running made no observable difference here — presumably
whether the JIT hoists the isRunning() read out of the loop differs between our
setups. That doesn't change anything about the fix being right; it just means the
old test's determinism rested on unspecified behaviour, as you said. Worth
mentioning in case the difference matters to you.

3. Tests. Replaced my test with your A and B. A uses a latch in the
acknowledgement executor to establish that the threshold flush completed before
the remainder is acked, and an interval that outlasts the shutdown timeout, so a
scheduled execution exists but provably cannot fire during the test. Both fail on
main and pass with the change.

Full module is green: 636 tests, 0 failures, 6 skipped, spotless checks enabled,
LocalStack integration tests included.

---

One thing I noticed while working through your "the polling thread can hold a
just-polled message" point, which I have deliberately left out of scope. The same
window is also reachable from the wait loop's exit condition:

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

In between, the message is in neither acks nor acksBuffer, so hasAcksLeft()
can report false, the wait loop exits, waitAcknowledgementsToFinish() sets
isTimeoutElapsed and clears the buffer, and the polling thread then adds the
message to a buffer nothing will flush. It is dropped with no warning at all,
unlike the timeout case. The window is narrow, but it widens under contention on
context.lock(). Happy to open a separate issue if you think it's worth
addressing.

[tomazfernandes]

Thanks for the issue and PR @hyeongguen-song.

This addresses both the sub-threshold remainder scenario you initially brought up, the interval = ZERO case, and makes running volatile to fix stale reads during shutdown.

Happy to open a separate issue if you think it's worth addressing.

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.