Backport SQS fixes to 3.4.x - #1671
Merged
Merged
Conversation
tomazfernandes
requested review from
MatejNedic and
maciejwalkowiak
as code owners
August 8, 2026 18:20
Messages left in the acknowledgement buffer below the acknowledgement threshold were never flushed on shutdown. The processor spun for the whole acknowledgementShutdownTimeout, logged Acknowledgements did not finish in 20000 ms. Proceeding with shutdown. and then cleared the buffer, so those messages were never deleted from SQS and were redelivered once the visibility timeout expired. Drain the buffer from the shutdown wait loop instead, which makes the flush a guarantee of the shutdown path itself rather than something that depends on a scheduled execution being armed. This also covers acknowledgementInterval = ZERO with a positive threshold, where no scheduled execution is ever created and a remainder was lost on every shutdown. The flush waits for the ack queue to drain so batches are not needlessly split, and it is retried on every iteration of the wait loop: 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. Also make running volatile in AbstractOrderingAcknowledgementProcessor. It is written under lifecycleMonitor but read without synchronization from the polling and scheduler threads, so those threads had no guarantee of observing stop(). Issue awspring#1661 (cherry picked from commit 40721da)
…wspring#1664) 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: Message<T> polledMessage = this.acks.poll(1, TimeUnit.SECONDS); if (polledMessage != null) { addMessageToBuffer(polledMessage); If the wait loop sampled in that window 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 shutdown timeout case, this dropped the message with no warning at all. Track the messages that have been received but are not in the queue nor the buffer yet, and include 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; the brief double counting while the message sits in the buffer only makes the wait more conservative. At most one message per shutdown can be in this window, but 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. Follow-up to awspring#1663. (cherry picked from commit 423f8b7)
…wspring#1632) * fix(sqs): prevent ConcurrentModificationException on container stop When an SQS message listener container is stopped, AbstractPollingMessageSource.stop() iterates over this.pollingFutures to cancel active polls. Each future has a whenComplete callback registered in managePollingFuture that synchronously/concurrently removes itself from the pollingFutures collection. Iterating over the collection while elements are being removed throws a ConcurrentModificationException. This commit prevents the exception by taking a shallow snapshot copy of the pollingFutures collection before iterating and cancelling the futures. Fixes awspring#1594 * test(sqs): add regression test for ConcurrentModificationException on container stop Cancelling an in-flight polling future during stop() runs the whenComplete callback registered in managePollingFuture inline on the stopping thread, removing the future from pollingFutures while the collection is iterated. The synchronizedCollection mutex is reentrant, so synchronization alone cannot prevent this; iterating a snapshot copy does. Regression introduced in awspring#1455 by replacing thenRun with whenComplete, first released in 3.4.1 and 4.0.0. --------- Co-authored-by: Tomaz Fernandes <tomaz.fernandes.se@gmail.com> (cherry picked from commit 131e941)
tomazfernandes
force-pushed
the
3.4.x-sqs-fixes
branch
from
August 8, 2026 19:34
7878b84 to
39d9478
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Cherry-picks the following SQS fixes from
mainonto3.4.x:ConcurrentModificationExceptionon container stopAll three are unmodified cherry-picks (
-x) of the original commits. Verified locally by running the test classes the picks touch.