BatchingAcknowledgementProcessor: buffered acknowledgements are never flushed on shutdown (remaining half of #925) #1661

Type: Bug

Component:
SQS

Describe the bug

TL;DR — When a container is stopped while acksBuffer holds 1 .. ackThreshold - 1 messages, those acknowledgements are never sent: shutdown blocks for the full acknowledgementShutdownTimeout (20s by default), the buffer is then cleared, and the messages are redelivered after the visibility timeout even though the listener completed them successfully. executeAllAcks() is reachable only from the scheduled flusher, which is already dead by then, so no unconditional final flush exists — the wait added by #1082 has nothing left to wait for. Workaround: acknowledgementInterval(Duration.ZERO).

Versions:

• spring-cloud-aws 4.1.0 (also present in current main, a1235192a54a9a19c0d19806e602509781a53bc0)
• Spring Boot 4.1.0, JDK 25
• Standard (non-FIFO) queue, AcknowledgementMode.ON_SUCCESS, acknowledgementInterval / acknowledgementThreshold left at the Standard-SQS defaults (1s / 10)

This is the remaining half of #925. PR #1082 made doStop() wait for the acknowledgement buffer to drain, and the threshold path can still drain full batches during that window — but nothing is left that can drain a sub-threshold remainder. Under sustained load the wait therefore reliably times out and those buffered acknowledgements are discarded instead of being sent.

AbstractOrderingAcknowledgementProcessor.stop() sets running = false before calling doStop(). From that moment on:

1. The scheduled (interval) flusher refuses to re-arm itself, so only the already-pending execution can still fire — after that the chain is dead:

   spring-cloud-aws/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/BatchingAcknowledgementProcessor.java

   Lines 467 to 471 in a123519

   	private void scheduleNextExecution(Instant nextExecutionDelay) {
   	if (!this.context.isRunning()) {
   	logger.debug("AcknowledgementProcessor {} stopped, not scheduling next acknowledgement execution.",
   	this.context.id);
   	return;

2. That single remaining execution only flushes if it passes this guard:

   spring-cloud-aws/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/BatchingAcknowledgementProcessor.java

   Lines 501 to 507 in a123519

   	private void pollAndExecuteScheduled() {
   	if (Instant.now().isAfter(this.context.lastAcknowledgement.plus(this.ackInterval))) {
   	List<CompletableFuture<Void>> executionFutures = this.context.executeAllAcks();
   	if (executionFutures.isEmpty()) {
   	this.context.lastAcknowledgement = Instant.now();
   	}
   	}

   Whether that last run flushes is a race rather than an unconditional failure. It is deterministically lost whenever a threshold flush bumped lastAcknowledgement (via doExecute) after the pending scheduled task had already been armed: the task still fires at its original time, which is now earlier than lastAcknowledgement + ackInterval, so the guard fails, nothing is flushed, and — per (1) — nothing is rescheduled. Under sustained load that is the normal state of affairs, and it also keeps happening during the shutdown window, because the buffering thread deliberately stays alive and keeps moving messages from acks into the buffer (shouldKeepPollingAcks()), running the threshold executor on each one. At low throughput, by contrast, the scheduled path itself drives lastAcknowledgement, so the final run passes the guard and flushes normally.

3. The only remaining drain path is the threshold executor, which requires >= ackThreshold messages in a group:

   spring-cloud-aws/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/BatchingAcknowledgementProcessor.java

   Lines 432 to 438 in a123519

   	private List<CompletableFuture<Void>> executeThresholdAcks() {
   	this.context.lock();
   	try {
   	logger.trace("Executing acknowledgement for threshold in {}.", this.context.id);
   	return this.context.executeAcksUpTo(this.ackThreshold, this.ackThreshold);
   	}
   	finally {

   Polling has already stopped, so no new messages arrive and the buffer can never reach the threshold again.

4. waitOnAcknowledgementsIfTimeoutSet() therefore spins for the full timeout, warns, and acksBuffer is cleared:

   spring-cloud-aws/spring-cloud-aws-sqs/src/main/java/io/awspring/cloud/sqs/listener/acknowledgement/BatchingAcknowledgementProcessor.java

   Lines 241 to 268 in a123519

   	private void waitOnAcknowledgementsIfTimeoutSet() {
   	if (Duration.ZERO.equals(this.ackShutdownTimeout)) {
   	logger.debug("Not waiting for acknowledgements, shutting down.");
   	return;
   	}
   	try {
   	var endTime = LocalDateTime.now().plus(this.ackShutdownTimeout);
   	logger.debug("Waiting until {} for acknowledgements to finish", endTime);
   	while (hasAcksLeft() || hasUnfinishedAcks()) {
   	if (LocalDateTime.now().isAfter(endTime)) {
   	throw new TimeoutException();
   	}
   	Thread.sleep(200);
   	}
   	logger.debug("All acknowledgements completed.");
   	}
   	catch (InterruptedException e) {
   	Thread.currentThread().interrupt();
   	throw new IllegalStateException("Interrupted while waiting for acknowledgements to finish");
   	}
   	catch (TimeoutException e) {
   	logger.warn("Acknowledgements did not finish in {} ms. Proceeding with shutdown.",
   	this.ackShutdownTimeout.toMillis());
   	}
   	catch (Exception e) {
   	logger.warn(
   	"Error thrown when waiting for acknowledgement tasks to finish in {}. Continuing with shutdown.",
   	this.parent.getId(), e);

Net effect: whenever the buffer holds 1 .. ackThreshold - 1 messages at shutdown, those messages are never deleted from SQS, shutdown is delayed by the full acknowledgementShutdownTimeout (20s by default), and the messages are redelivered once the visibility timeout expires — even though the listener completed them successfully.

Observed in production (Kubernetes, standard queue, maxConcurrentMessages 10, maxMessagesPerPoll 10):

WARN io.awspring.cloud.sqs.listener.acknowledgement.BatchingAcknowledgementProcessor
     Acknowledgements did not finish in 20000 ms. Proceeding with shutdown.

The load dependence matches the mechanism above. This service never logged the warning while per-pod throughput was low: at that rate the scheduled path (not the threshold path) drives lastAcknowledgement, so the final scheduled run passes the guard and flushes normally. After we reduced the replica count, per-pod throughput roughly tripled and the warning started appearing.

This is particularly relevant for maxReceiveCount = 2 setups (as reported in #925): a message stranded on two consecutive deliveries goes to the DLQ despite having been processed successfully both times.

Sample

I don't have a standalone reproducer application, but the condition is deterministic and should be expressible as a unit test alongside the existing BatchingAcknowledgementProcessorTests. Note that the shutdown tests added by #1082 (testAckOnShutdown) don't cover it: they use acknowledgementInterval = Duration.ZERO, so the scheduled flusher never runs at all, and they acknowledge exactly 100 messages — an exact multiple of the threshold, leaving no remainder.

1. Configure acknowledgementInterval = 1s, acknowledgementThreshold = 10, acknowledgementShutdownTimeout = 20s.
2. Acknowledge exactly 10 messages, so the threshold executor flushes and sets lastAcknowledgement = now.
3. Immediately acknowledge N more messages with 1 <= N <= 9, and let them reach acksBuffer.
4. Call stop() within ackInterval of step 2.

Expected: the N messages are acknowledged — there is 20s of budget available.
Actual: hasAcksLeft() stays true, the wait times out after 20s, and the N messages are dropped by acksBuffer.clear().

Possible fix direction: let the scheduled execution keep flushing (or perform one final unconditional executeAllAcks()) while waitOnAcknowledgementsIfTimeoutSet() is still within its budget — doStop() only disposes the taskScheduler after that wait returns, so the scheduler is still usable for exactly this window. The buffering thread is already deliberately kept alive for the same window via shouldKeepPollingAcks() — arguably the flusher should be too, otherwise the wait added by #1082 has nothing to wait for.

Workaround for anyone hitting this: set acknowledgementInterval(Duration.ZERO), which selects ImmediateAcknowledgementProcessor. Thanks to the null threshold check at StandardSqsComponentFactory#L70, setting the threshold as well is not necessary.