Explain SqsListener concurrency

[andrej-urvantsev]

My use case is quite simple - read messages from SQS, transform them and write into relational database(jdbc).

Problem is that with all defaults, SqsListener starts 10 threads(I guess it's defined by maxConcurrentMessages) and therefore size of batches being read from SQS is a quite small - it's very rare when it's more than 1 message.

But 10 threads writing into DB one record take 10 DB connections. And writing just one record per connection is less than optimal.

So, I would prefer less concurrent threads for SqsListener, but bigger batches(even if it takes more polls). And here I struggle to understand how things work.

I would assume that logical thing to do is to set maxConcurrentMessages=2 and messagesPerPoll=30, but that is not possible, application fails with messagesPerPoll should be less than or equal to maxConcurrentMessages, because maxConcurrentMessages meaning is actually "max inflight messages".

So, how do I maximize batch size and minimize concurrency?

[maciejwalkowiak]

@lazystone it looks like batching consumer would fit your use case well. Have you had a look at https://docs.awspring.io/spring-cloud-aws/docs/3.2.0/reference/html/index.html#batch-processing ?

[andrej-urvantsev]

I did a small experiment: I stopped consumer, to accumulate some events in SQS(test env - low flow).

I tried to enable BATCH mode explicitly, just to check is there any difference:

    @Bean
    MessageListenerContainer<String> listenerContainer(
            SqsAsyncClient sqsAsyncClient,
            @Value("${application.incoming-events-queue}") String queueNames,
            @Value("incoming_message_${INSTANCE_ID:${random.value}}") String id,
            IncomingEventsMessageListener incomingEventsMessageListener) {
        return SqsMessageListenerContainer.<String>builder()
                .configure(options -> options.listenerMode(ListenerMode.BATCH))
                .id(id)
                .sqsAsyncClient(sqsAsyncClient)
                .queueNames(queueNames)
                .messageListener(incomingEventsMessageListener)
                .build();
    }

Then I started it, there were 55 messages in SQS and logs looked like this(timestamp, thread name, batch size):

2024-09-19 14:00:26	incoming_message_subscription-webhook-sender-5db94c979d-h4948-1: incoming-events-queue batch size: 10
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-2: incoming-events-queue batch size: 10
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-3: incoming-events-queue batch size: 4
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-4: incoming-events-queue batch size: 5
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-5: incoming-events-queue batch size: 5
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-6: incoming-events-queue batch size: 6
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-7: incoming-events-queue batch size: 10
2024-09-19 14:00:27	incoming_message_subscription-webhook-sender-5db94c979d-h4948-8: incoming-events-queue batch size: 5

I would expect to see batch sizes like "10, 10, 10, 10, 10, 5"

Disclaimer: I'm very new to SQS world, so maybe I'm missing something obvious...

[maciejwalkowiak]

As far as I can tell there is no guarantee you'll always get 10 messages as they may come from different SQS shards (?). You can repeat similar experiment with pure SQS client - as far as I remember when I did that few years ago the results were the same.

[andrej-urvantsev]

Ok, I did a couple of more experiments and it seems that behavior depends of how many events are visible at the moment in SQS - more events in the queue, more consistent batches of 10.

Then the last question remain: I understand that it should be possible to configure batching somehow, that it does multiple polls, combine results and provides it to the consumer.

Does that mean that if I do something like this:

    @SqsListener(
            queueNames = {"${application.incoming-events-queue}"},
            id = "incoming_message_${INSTANCE_ID:${random.value}}",
            maxConcurrentMessages = "20")
    public void handle(List<Message<String>> messages) {

and if there are enough messages visible in SQS, then SQS listener container will do 2 polls and invoke handle method with 20 messages?

[tomazfernandes]

Hi @lazystone, you're correct, if you set maxConcurrentMessages for a value greater than 10 it'll combine the result of multiple polls.

And as @maciejwalkowiak pointed out, that means you'll get messages up to that amount, but SQS does not guarantee to always deliver that amount.

And as you found out, the more messages in SQS, the higher the chances you'll receive full batches.

You don't need to manually configure BATCH mode - just having the List parameter will configure that for you.

[andrej-urvantsev]

Thanks for clarification!