How do I use the messagegroupid parameter to only receive queue messages tagged with the id I need?
I've been trying to use the line below to retrieve but it will always receive all the queue messages from other group id as well.
List<Message> messages = sqs.receiveMessage(receiveMessageRequest.withAttributeNames("MessageGroupId")).getMessages();
What should be the correct way to do it?
The ReceiveMessageRequest is not used for filtering based on message attributes. If you look at the docs for ReceiveMessageRequest.html.withAttributeNames() it says:
A list of attributes that need to be returned along with each message.
In general, you can't filter on the messages you get back from SQS. You can limit the number but you can't say, for example, "give me all the messages that match this pattern".
My solution was to utilize the ChangeMessageVisibilityBatchRequest (see docs) which essentially sends messages back to the queue to be re-processed.
My lambda has a time-based trigger. Each time it turns on I receive messages until there aren't any more. For each batch I group the messages by MessageGroupId, process and delete the first group, and send the remaining group messages back to the queue to be picked up during the next iteration.
Here's the gist of my code:
(Note the _awsSqsService.SendBackToQueueAsync(groupMessages) method ultimately sends the messages back to the queue via the AWS SQS ChangeMessageVisibilityBatchRequest)
public async Task Run()
{
var batchContainsMessages = true;
while (batchContainsMessages)
{
var messageBatch = await _awsSqsService.GetMessageBatchAsync();
if(messageBatch.Messages.Count > 0 && messageBatch.HttpStatusCode == HttpStatusCode.OK)
{
await ProcessMessageBatchAsync(messageBatch.Messages);
}
else
{
batchContainsMessages = false;
}
}
}
private async Task ProcessMessageBatchAsync(List<Message> messages)
{
// SQS fifo queues will often return a batch of messages with different MessageGroupIds.
// Because of this, we need to group them ourselves, process one group (the first group),
// and send the rest back to the queue to be processed in the next iteration.
// This ensures that we process as many messages as possible per group in a single batch (max is 10)
var messageGroups = GetMessageGroups(messages);
var isFirstGroup = true;
foreach (var group in messageGroups)
{
var groupId = Int32.Parse(group.Key);
var groupMessages = group.Value;
if (isFirstGroup)
{
isFirstGroup = false;
await ProcessMessagesAsync(groupId, groupMessages);
await _awsSqsService.DeleteMessagesAsync(groupMessages);
}
else
{
await _awsSqsService.SendBackToQueueAsync(groupMessages);
}
}
}