Add support for FIFO queues

[TheFarmer1]

When receiving messages from FIFO queues in batches, the handleMessage function is called in parallel for each message in the batch. For FIFO queues, this should be serially processed to maintain the correct order of processing. Ideally, the MessageGroupID could be used to process messages in different groups in parallel.

If batchSize is set to 1, then FIFO queues work ok naturally.

[nspragg]

We're not looking to support FIFO queues at this time.

[tnolet]

@TheFarmer1 @nspragg I'm working on a PR to add MessageGroupID based handling of FIFO queues. Not that much actually, as far as I can see now.

The speed gains are pretty good though.

[OrKoN]

@tnolet did you figure out what is needed to support FIFO queues?

[tnolet]

@OrKoN I think so, will need to run some production load for a bit still. Don’t have a clear estimate.

[robbie-thompson]

@nspragg Still no plans to support FIFO?

[VrushabhGore]

@tnolet Did you do something for this?

[SAGARACH65]

@tnolet Any update on this.?

[matiasgarcia]

sqs-consumer supports FIFO out of the box, you just need to know how to use it in conjunction of your sqs queue configuration.

One option would be setting batchSize: 1, although this would be a terrible idea.

Another option is using:

• batchSize > 1
• handleMessageBatch
• visibilityTimeout
• heartbeatInterval

As long as you configure these correctly, you shouldn't have major issues.

Here is an snippet:

const app = Consumer.create({
  queueUrl: "http://localhost:4599/000000000000/fifo-queue.fifo",
  batchSize: 10,
  waitTimeSeconds: 0,
  /*
    Use visibilityTimeout in conjunction of heartbeatInterval to avoid assigning an on-going message group to another consumer
  */
  visibilityTimeout: 5,
  heartbeatInterval: 2,
  handleMessageBatch: async (messages) => {
    const messagesToAcknowledge = [];
    try {
      for (const message of messages) {
        await sleep(1000 + rand()); // simulate different workload
        console.log('Done consuming message:', message.Body, new Date().toISOString())
        messagesToAcknowledge.push(message);
      }
      return messagesToAcknowledge;
    } catch (err) {
      // Processing one message failed, so we want to abort and avoid processing messageGroupsMessages.
      console.error('messaging processing failed', err);
      throw err;
    } finally {
      return messagesToAcknowledge;
    }
  },
  sqs,
});