Spring-cloud-aws: Enhancement: Add possibility to configure maxNumberOfMessages per SqsListener

[arseniiR]

Scenario: I have 2 queues, one 'my-queue-1' of type = standard queue and 'my-queue-2' of type FIFO queue. I would like to grab as many messages as possible from 'my-queue-1', and want to preserve order during message processing on 'my-queue-2' by consuming only one message at a time.
For example, for 'my-queue-1' it ca be 10 messages, but for 'my-queue-2' it can be 1.

Proposal: SqsListener#maxNumberOfMessages may be optional and when specified it will override
SimpleMessageListenerContainer#setMaxNumberOfMessages().
This will permit to configure the number of messages in batching for each queue.

[bisoldi]

I've been trying to figure out how to create and set a different SimpleMessageListenerContainer per @SQSListener, which would include setting just the maxNumberOfMessages. My use case is to have 2 listeners, each requiring a different AmazonSQSAsync (one requiring STS credentials and another requiring default credentials). If there is a way to do that, I would certainly like to know how.

By the way, your FIFO queue, by definition, preserves order. I would think that even if you request multiple messages, they are provided back to the container in order and therefore should be passed to the listener in order. Is it not doing that?

[arseniiR]

As I can see, the maxNumberOfMessages is added for each request to receive messages.
See: com.amazonaws.services.sqs.model.ReceiveMessageRequest#withMaxNumberOfMessages
and:
org.springframework.cloud.aws.messaging.listener.AbstractMessageListenerContainer.QueueAttributes#getReceiveMessageRequest

Thus, you don't need to add different SimpleMessageListenerContainer per @SQSListener! This means you can use the same client.

About FIFO queue, indeed it preserve order. Still this order is preserved per group. This is very important to understand. Once a consumer will get a message from a group, other consumers will not be able to take messages from that group (until inflight message is either deleted or visibility timeout expire), still other consumers can consume messages from other groups.
Now, if I have a consumer that read a message from group-1, and delegate message to other thread to process it, that consumer can go and take next message from other group....

[bisoldi]

Not sure if you are intentionally grouping the messages, as you don't need to.

From the docs:
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html
Every message sent to a FIFO queue requires a message group ID. If you don't need multiple ordered message groups, specify the same message group ID for all your messages.

[arseniiR]

Sometimes you need to use multiple groups. For example to process, in order, messages for a customer. If you have many customers, for each customer you will have one group. This will ensure fairness and will allow message processing in a concurrent way across customers.
Of course if you want to preserve order across all the messages in the queue, then you need to use one group. In this case, once a consumer will retrieve a message (or more than one message - i.e. message batching), then other consumers will wait for the entire group to be unlocked (i.e. the group is locked while there is at least one in-flight message in the group) - since there is only one group, consumers are forced to wait for all in-flight messages to be processed (i.e. either to be deleted or visibility timeout expires).

[pkurczyna]

I've been trying to figure out how to create and set a different SimpleMessageListenerContainer per @SQSListener, which would include setting just the maxNumberOfMessages. My use case is to have 2 listeners, each requiring a different AmazonSQSAsync (one requiring STS credentials and another requiring default credentials). If there is a way to do that, I would certainly like to know how.

By the way, your FIFO queue, by definition, preserves order. I would think that even if you request multiple messages, they are provided back to the container in order and therefore should be passed to the listener in order. Is it not doing that?

this would be great feature to have

[zg-yyl]

I've been trying to figure out how to create and set a different SimpleMessageListenerContainer per @SQSListener, which would include setting just the maxNumberOfMessages. My use case is to have 2 listeners, each requiring a different AmazonSQSAsync (one requiring STS credentials and another requiring default credentials). If there is a way to do that, I would certainly like to know how.

Have you had any luck figuring out how to configure different SimpleMessageListenerContainers per listener?
My alternate approach has been to use https://github.com/awslabs/amazon-sqs-java-messaging-lib instead - it allows us to specify the containerFactory for each @JmsListener

[sushanted]

Scenario: I have 2 queues, one 'my-queue-1' of type = standard queue and 'my-queue-2' of type FIFO queue. I would like to grab as many messages as possible from 'my-queue-1', and want to preserve order during message processing on 'my-queue-2' by consuming only one message at a time.
For example, for 'my-queue-1' it ca be 10 messages, but for 'my-queue-2' it can be 1.

Proposal: SqsListener#maxNumberOfMessages may be optional and when specified it will override
SimpleMessageListenerContainer#setMaxNumberOfMessages().
This will permit to configure the number of messages in batching for each queue.

Was facing the same issue

Following 'hack' worked for me:

`@Bean
public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSqs) {

    return new SimpleMessageListenerContainerFactory() {
        @Override
        public SimpleMessageListenerContainer createSimpleMessageListenerContainer() {
            SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer() {
                @Override
                protected void startQueue(String queueName, QueueAttributes queueAttributes) {

                    // A place to configure queue based maxNumberOfMessages

                    try {
                        if (queueName.endsWith(".fifo")) {
                            FieldUtils.writeField(queueAttributes, "maxNumberOfMessages", 1, true);
                        }
                    } catch (IllegalAccessException e) {
                        throw new RuntimeException(e);
                    }
                    super.startQueue(queueName, queueAttributes);
                }
            };
            simpleMessageListenerContainer.setAmazonSqs(amazonSqs);
            return simpleMessageListenerContainer;
        }
    };
}`

[lenlaurito]

Is it possible to configure different maxNumberOfMessages for listeners of different standard queues?

[maciejwalkowiak]

@lenlaurito there is a global configuration option. There is no option per-listener. What's your use case?

[lenlaurito]

@maciejwalkowiak Our application has listeners with different needs. Some needs to have lower max number of messages because it might take a while to finish and it might need more resources (CPU/memory). While other listeners have processes that are quick and don't require much resources and these listeners usually need to process a lot of messages so having a higher max number of messages would be ideal.

[maciejwalkowiak]

Got it