spring-cloud-aws
spring-cloud-aws copied to clipboard
maxDelayBetweenPolls is not honored if there is long-running SQS task during LOW throughput mode
Type: Bug
Component: SQS Version: 3.0.3 Config:
- maxDelayBetweenPolls: 10000
- maxConcurrentMessages: 121
- listenerShutdownTimeout: 120000 All other portions of the config are the defaults.
Describe the bug
Bug Description
Hi team,
My application uses @SqsListener
to process messages from an SQS queue. Occasionally, it experiences an issue where no SQS messages are received or processed for several minutes, despite thousands of messages being present in the queue. I enabled DEBUG
logs for io.awspring.cloud.sqs
to investigate this problem.
During the last time the issue occurred, I observed the following log every 10 seconds (but no polling logs):
Trying to acquire full permits for io.awspring.cloud.sqs.sqsListenerEndpointContainer#4-0. Permits left: 120
Additionally, I could see logs in my application indicating that a single SQS message was actively being processed. As soon as that task finished processing, polling started up again and there were plenty of logs like:
Polling queue <MY-QUEUE> for 10 messages.
Bug Scenario
After looking through the awspring source code, I believe the following sequence of events occurs:
- awspring is in
CurrentThroughputMode.HIGH
mode - An SQS message is received that will take several minutes to process
- While this SQS message is processing, awspring attempts to poll the queue for 10 messages, but no messages are received. This causes
SemaphoreBackPressureHandler
to enterCurrentThroughputMode.LOW
mode. - In the
AbstractPollingMessageSource.pollAndEmitMessages()
loop, whenSemaphoreBackPressureHandler.requestInLowThroughputMode()
is run, it tries to acquire all 121 permits. However, one permit remains unavailable due to the long-running SQS task that started earlier. BecauserequestInLowThroughputMode
cannot acquire all permits, it acquires none, resulting in no SQS polling. - Once the long-running SQS task completes, all permits are available, and polling begins again.
For step 5, I see the following logs:
...
Processed SQS notifications message (this log is from my own app)
Releasing permit for queue <MY-QUEUE>
1 unused permit(s), setting TM HIGH for io.awspring.cloud.sqs.sqsListenerEndpointContainer#4-0. Permits left: 120
Acquired full permits for io.awspring.cloud.sqs.sqsListenerEndpointContainer#4-0. Permits left: 0
Polling queue <MY-QUEUE> for 10 messages.
...
I believe this happens because SemaphoreBackPressureHandler.requestInLowThroughputMode()
uses the following code. With this approach, LOW
can only acquire permits if the full totalPermits
are available. While this prevents parallel SQS polling in LOW
mode, it also prevents any polling if there ongoing SQS tasks consuming permits. With fast processing of SQS messages, this would not be noticeable, but I occasionally have SQS tasks that last several minutes.
private int requestInLowThroughputMode() throws InterruptedException {
// Although LTM can be set / unset by many processes, only the MessageSource thread gets here,
// so no actual concurrency
logger.debug("Trying to acquire full permits for {}. Permits left: {}", this.id,
this.semaphore.availablePermits());
boolean hasAcquired = tryAcquire(this.totalPermits, CurrentThroughputMode.LOW);
if (hasAcquired) {
logger.debug("Acquired full permits for {}. Permits left: {}", this.id, this.semaphore.availablePermits());
// We've acquired all permits - there's no other process currently processing messages
if (!this.hasAcquiredFullPermits.compareAndSet(false, true)) {
logger.warn("hasAcquiredFullPermits was already true. Permits left: {}",
this.semaphore.availablePermits());
}
return this.batchSize;
}
else {
return 0;
}
}
Expectation
I would expect that if my application is handling an SQS message that takes several minutes to process, and SemaphoreBackPressureHandler
is in LOW
mode, then SqsListener
would continue to poll the queue for up to 10 messages every 10 seconds.
Workaround
As a workaround, I am using backPressureMode=FIXED_HIGH_THROUGHPUT
to prevent my application from entering LOW
mode. However, this results in additional polling.
Please let me know if there is any additional information I can provide or if any of my assumptions are incorrect.