fix(sqs): prevent ConcurrentModificationException on container stop

[Baqirrizvidev]

When an SQS message listener container is stopped under heavy load, AbstractPollingMessageSource.stop() iterates over this.pollingFutures to cancel active polls. Each future has a whenComplete callback registered in managePollingFuture that removes itself from the pollingFutures list.

Cancelling the future inside the iteration loop triggers the callback synchronously/concurrently, modifying the list while it is being iterated, which throws a ConcurrentModificationException and halts proper container shutdown.

This commit resolves the issue by taking a snapshot copy (new ArrayList<>(this.pollingFutures)) of the collection before iterating and cancelling, decoupling iteration from list modifications.

Fixes #1594

[tomazfernandes]

Thanks for the PR @Baqirrizvidev.

As explained in the original issue, the new ArrayList() approach has the same problem of throwing ConcurrentModificationException when modifications happen while the new list is being built, and further synchronization is required.

Let me know if you'd like to amend the PR.

[KimDoubleB]

I'm hitting this one too, so I've been waiting on this PR.

I had a look at the concern raised about the snapshot:

the new ArrayList() approach has the same problem of throwing ConcurrentModificationException when modifications happen while the new list is being built

I might be missing something, but I don't think that applies to this particular field. pollingFutures is a Collections.synchronizedCollection, and new ArrayList<>(Collection) copies through c.toArray(), which the synchronized wrapper already runs under the mutex — so the copy itself looks atomic to me as it stands.

The way I read ConcurrentModificationException, it isn't really a cross-thread race either: cancel(true) completes the future, the whenComplete callback registered in managePollingFuture then runs inline on that same thread, and it removes from the very collection being iterated. If that's right, iterating a private copy is the part that actually fixes it, and the mutex isn't the load-bearing bit here.

Not arguing against the explicit synchronized block — it documents the intent and would survive the field's type changing later. I just wondered whether the one-liner might already be enough on its own, in case that's what's holding this up.
Happy to be told I've got this wrong.

[tomazfernandes]

@KimDoubleB I recall trying the new ArrayList() approach and getting ConcurrentModificationException, but it's been a while. I'll take another look.

[tomazfernandes]

Thanks @Baqirrizvidev for the fix, and @KimDoubleB for the analysis, which I can confirm is correct: the snapshot copy is atomic, since the synchronized wrapper runs toArray() under its mutex, and the exception comes from the whenComplete callback removing the future inline on the stopping thread during cancellation.

I've rebased the branch on current main, dropped the unrelated ConfigurationChangeDetector commit, and added a regression test. I'll merge the PR as soon as the tests return green.

[tomazfernandes]

Backported to 3.4.x in #1671