Maintaining order in FIFO queues

Hi @JaidenAshmore ,

Great work on this library, it is very well done.

I was wondering if you have worked on any solutions for FIFO queues? Basically we would like to process all messages that were pulled in order, so the current concurrency does not work as it breaks the order of the messages.

For example:

    @PrefetchingQueueListener(
            value = "${sqs.queue}",
            concurrencyLevelString = "10",
            desiredMinPrefetchedMessages = 10,
            maxPrefetchedMessages = 20)
    public void processSQS(Message message, Acknowledge ack) {

This code, If I understood everything correctly, will grab 10 messages and distributes those 10 into the 10 concurrency threads, which would break the ordering.

Would it be possible to have something like:

    @PrefetchingQueueListener(
            value = "${sqs.queue}",
            concurrencyLevelString = "10",
            desiredMinPrefetchedMessages = 10,
            maxPrefetchedMessages = 20)
    public void processSQS(List<Message> messages, Acknowledge ack) {

So this would grab 10 messages from the queue and pass those on to be processed as a group, maintaining the order. Not sure how the Acknowledge would work on this case, if it should be per message or per batch.

Again, thank you for the great work.

[JaidenAshmore]

Hey, thanks for the interest!

Yeah your understanding of the API is correct. Hmmm...this is an interesting use case and I can see the reason for wanting this feature.

One problem I can see is that it wouldn't work in a multi node service. Just because this node grabs 10 messages in order doesn't mean that a different node will grab another 10 and process them in the correct order compared to that original node. Were you only planning on running your app on a single node?

Let me ponder on this, I will need to do a bit of research on how SQS recommends this approach or whether they just recommend to only process on a single node. Based on that I can figure out whether that is something I would be able to integrate into the library.

A workaround for you would be to just run it on a single and node and set the concurrency level as 1. This is unfortunate if your system is I/O heavy and have a lot of messages but if you aren't processing a lot then this would be enough for now.

Hi @JaidenAshmore , thanks for the super quick reply!

I can only speak about AWS FIFO implementation, as that is what we are using, but in that case it would work even in multi nodes. The AWS implementation uses a groupID to maintain the order, so if a node grabs messages from groupID 1, say, then that same groupID is made invisible to another node. This whole process should be invisible to the consumer, so if my understanding of the Java Dynamic SQS Listener library is correct, the change needed would be passing the whole batch to the final processing method, instead of only a single message.

Here is more info: https://aws.amazon.com/sqs/faqs/#:~:text=By%20design%2C%20Amazon%20SQS%20FIFO,message%20groups%20to%20different%20consumers.

https://aws.amazon.com/blogs/developer/how-the-amazon-sqs-fifo-api-works/

We thought about the workaround but we think it might affect the performance too much unfortunately.

Anyways, thank you very much for taking this into consideration!

[JaidenAshmore]

Perfect thanks for the links. Let me see what I can do, will keep you posted.

Awesome, thank you very much!

[JaidenAshmore]

Okay sorry for the delay, I have been letting this process in the background to see what we could do.

So ultimately all we we need to guarantee is that all messages with the same message group id are not processed at the same time within that node. As the SQS Client should be able to make sure that two consumers (nodes) don't get messages from the same group we don't need to worry about cross node concurrency problems. What I am thinking is that we may not actually need to get the message listener to allow a list of messages (which unfortunately would be difficult with this framework). If we can make sure that the underlying framework does the effort to guarantee that two messages with the same group ID do not run at the same, we can maintain the existing library like:

public void processSQS(Message message, Acknowledge ack) {

Design

API

@FifoMessageListener("queueName.fifo")
public void messageListener(@Payload String messageBody) {

}

Potential core implementation Details

Currently the concurrency of message processing is completed using the ConcurrentMessageBroker. This will take the message from the MessageRetriever, place it onto a thread in its thread pool which calls the MessageProcessor with it for processing. This thread concurrency logic doesn't do anything smart so what we could do is create a new MessageBroker that is FIFO compliant.

This FIFO compliant MessageBroker could contain a thread safe Set of the message group IDs currently being processed and not allow processing of another message until that is finished.

There could be difficulties with making sure we are getting true concurrency with this approach. For example, if we want a concurrency of 10 but the MessageBroker downloads 10 messages of the same group ID, what would be the best way to know to download more without constantly pulling more and more messages which unfortunately are the same GroupID.

Question

Do you think that this design without supporting a list of messages would cover your use case? If you need it to provide a list of messages I have a different workaround that may work for you.

Hi @JaidenAshmore , not at all, that was really quick!

I believe your design would work and it looks really clean to control the processing order with a Set in the MessageBroker but I think that may indeed be a problem for concurrency. In my case it may be fine as I put messages in the queue alternating groupIDs, but I believe the usual case is to put all messages from one groupID then put all messages from another groupID, more or less like this case: https://stackoverflow.com/questions/62666518/connecting-multiple-producers-and-multiple-consumers-using-fifo-aws-sqs-with-mes.

If one does put all messages for the same groupID in the queue together, it may be that you will pull from the queue lots of times before getting a mix of groupIDs, which would make the concurrency of the consumer almost non-existent with this approach. Since FIFO queues are, by design, slower to consume, losing this little bit of concurrency might be a complicated drawback.

The List solution I thought of was to make use of the invisible function to allow you to pull multiple times from the queue, since you know that each batch of 10 will never be from the same groupID. That way you could pass 10 messages each to 10 threads and they could work in true concurrency, as you can be sure that each is working with a separate groupID.

One thing that occurs to me as I am writing this now, is that maybe the MessageBroker could work on the concurrency per batch of messages instead of per groupID, making use of the invisible function. So for each thread, it pulls a batch of 10 messages. Each batch will for sure be for a different GroupID than all the others and the messages could be a mix of groupIDs or the same, it does not mater, since they will be processed in the order they have been retrieved from the queue, one by one. The MessageBroker would be responsible for controlling that each thread consumes each batch in the correct order and you would still have some concurrency.

The MessageBroker would then maintain something like: Map<ThreadId, List<Messages>>

This would give you basically the same functionality as my list idea, but without the use of an actual list. What do you think?

[JaidenAshmore]

Hey mate, just to keep you updated, I am still looking at this during the time I have off. I am currently wrangling with how to get the concurrency working well and will keep you posted when I have something semi workable.

Awesome @JaidenAshmore , thanks for the update! Let me know if I can help with anything. As I have to solve this problem anyways for a project, we might as well tackle it together!

[JaidenAshmore]

Okay, I think I have something generally working however I need to do a lot of code cleanup, more manual testing, update documentation and write tests.

If you are interested, the current WIP changes are here: PR#325.

If you want to muck around with it you can clone this repo, checkout that branch and run the example spring boot app at examples/fifo-example.

Let me know if you want a SNAPSHOT release to do your own testing. I can create one from this branch pretty easily.

Hey, that is awesome! Will take a look later today!

Did you manage to work around the concurrency issue when all the messages from the same pull have the same group?

I think I can wait for the release, no need for a SNAPSHOT so as to not bother you so much. Thanks a lot!

Hi @JaidenAshmore , I was doing some tests and it looks like even if you put messages interchanging the groupIds into the queue, the queue will try to aggregate the groupIds when you do a pull:

When receiving messages from a FIFO queue with multiple message group IDs, Amazon SQS first attempts to return as many messages with the same message group ID as possible. This allows other consumers to process messages with a different message group ID.

(https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/FIFO-queues.html)

So it looks like that if you have multiple consumers/threads pulling from the queue at the same time, each one will most likely get a different groupId, and only if there are less than 10 in the same groupId is that the same batch could have mixed groups.

Not sure how that affects the concurrency in your implementation.

Otherwise, I have looked at the PR and it looks good to me!

The idea I have implemented before we talked was pretty crude, I just did a loop that maintains X threads always working. Each worker will run the ReceiveMessageRequest code to pull 10 messages and process them, deleting after it.

        while (!Thread.currentThread().isInterrupted()) {
            if (taskExecutor.getActiveCount() < workersSize) {
                QueueConsumerWorker worker = new QueueConsumerWorker(queueUrl);
                taskExecutor.execute(worker);
            }
        }

Cheers and great work!

[JaidenAshmore]

Hey, yeah concurrency is pretty solid and am able to consistently process messages at the desired rate. The only concern I have at at the moment was that I am sometimes seeing occurrences of say group 10 processing 7 messages over the time period but group 16 only has processed 1 even though both message groups have sent up the same number. I am just processing what is coming from the SQS server so I am hoping it is just a weird feature in ElasticMQ or I am accidentally de-duping messages by not providing a unique de-deplucation ID when I am sending the messages. Will look into that next.

even if you put messages interchanging the groupIds into the queue, the queue will try to aggregate the groupIds when you do a pull

Yep, in my FIFO example I have a scheduled job that is sending a message batch with each message having a unique group ID from 1 - 20. I then send another batch 10 messages all the same message group ID (21). In this scenario they are all getting processed in the correct order and are not executing two messages in the same group at the same time.

In MessageListeners I have defined the listener like:

@FifoListener(value = "test-queue.fifo", concurrencyLevel = 10, batchSize = 2)

The batch size is the maximum number of messages from to pull down at the same time. This listener will request messages while there is still available concurrency, placing them into a Map from message group ID to the list of cached messages in that group. The executor will then look into the map for a message group that isn't currently running and execute a message from that cache. So in this case we still get the full concurrency and can handle message receive calls that are in different groups.

Will keep you posted :)