Eventbridge rule with SQS fifo queue as target

[pedroTassinari]

Hi guys!

I would like to know if it's possible to set messageGroupId dynamically based on some property received from eventBridge event.

I've tried something like

new targets.SqsQueue(this.queue, {
  deadLetterQueue: this.queueDLQ,
  messageGroupId: events.EventField.fromPath('$.detail.data.id'),
});

but it didn't work.

Really appreciate any help!

[photz]

The docs don't mention that it can be dynamic, so it's probably not possible:
https://docs.aws.amazon.com/eventbridge/latest/APIReference/API_SqsParameters.html

For what it's worth, you can always have a Lambda in-between which sets the message group ID.

[royzhang107]

[CoreySchnedl]

messageGroupId: events.EventField.fromPath('$.detail.data.id')

This does not work as expected. Tested thorougly.

[BardiaN]

We’ve been having this issue since longtime, I’m wondering why AWS doesn’t fix this problem. It’s basically rendering this functionality useless. I’m thinking about switching to SNS to fix this issue.

[RichardHudsonCodes]

The only way that I can see of doing this is to create another SQS queue connected to a Eventbridge Pipe or Lambda which then adds a messageGroupId before passing it onto my target FIFO queue.

[eFuller96]

I have it working. Hope this helps someone as I spent so long looking, and reading conflicting sources on the internet.

Source is dynamodb Stream, and in my event bridge pipe I am able to set the following for my target SQS MessageGroupId:
$.dynamodb.Keys.PK.S

some aws docs that helped me: https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-input-transformation.html

[BardiaN]

I just retested it, and no it doesn't work, it sends the string literally to the SQS and the lambda I had which is not what we want.

Here is the cloud watch log:

{
    "ApproximateReceiveCount": "1",
    "SentTimestamp": "1726261510900",
    "SequenceNumber": "18888667020499951616",
    "MessageGroupId": "$.SK", // also tried "$.detail.SK", same result
    "SenderId": "AIDAISOKQBV44VMOZSBAU",
    "MessageDeduplicationId": "926c0c1f141ecfe2361c8865b1ed71e2c7a0b668e7fd6119241aa65aba7693ee",
    "ApproximateFirstReceiveTimestamp": "1726261510900"
}

I also wondered if it might be applied but we can't just see in the log attributes.
So I sent 20 messages simultaneously in a way that the messageGroupId would be different for each, so the messages would be treated in parallel, but no, I only had a single lambda execution, all messages were treated with the same group id.

The only thing I haven't checked yet is the same setup only with EventBridge Pipe, maybe you are talking about that 🤷‍♂️

[ianbrumby]

@BardiaN, I would recommend EventBridge Pipes if you can. This is definitely working for me now and shows up in my pipe configuration:

[BardiaN]

@ianbrumby Thanks for the recommendation but whether to use the rule or pipe depends deeply on the architecture.

1. Pipe doesn't unmarshal by default and that's something we need absolutely.
   a. yes, it accepts a lambda to transform but that'd be more expensive than just having a lambda as dynamodb stream handler because now you have to pay for both function and the pipe.
2. This GitHub issue is about EventBridge Rule and not EventBridge Pipe.

I hope AWS corrects this problem also on EventBridge rule

And finally, your image doesn't show if it really works or not, it could be any text in the MessageGroupId, how do you know for sure that it works ? (Do you see a rendered MessageGroupId in the target logs ? Or do you see a different execution in parallel for different MessageGroupIds ?)

[ianbrumby]

@BardiaN, understood. All good points, Apologies as I actually came to this thread through a EventBridge Pipe issue I was having and hadn't appreciated it was for standard rules. My bad.

And yes, I do the unmarshalling as part of enrichment phase of the pipe. I appreciate this will not suite everyone's architecture, however I'm not sure why you would need an additional Lambda function. The pipe will take the stream output directly and pass to a single function that can handle the unmarshalling and then pass to the queue. You can filter the DynamoDb stream in either method to reduce your Lambda calls.

If anyone does come here for Pipes, here is the rendered MessageGroupId in my logs, taken from the unmarshalled output of the enrichment phase of the pipe:

"attributes": {
      "ApproximateReceiveCount": "1",
      "SentTimestamp": "1726567479896",
      "SequenceNumber": "387823626822753959936",
      "MessageGroupId": "ORDER#01J7ZQ2Z5YDSPNQ432RPV",
      "SenderId": "xxxxx",
      "MessageDeduplicationId": "9ab7b450df7024453c49bca11522e785d2cf9822df0baee65720bf3cfc8142fa",
      "ApproximateFirstReceiveTimestamp": "1726567479896"
  },

[BardiaN]

Awesome, thanks for the log @ianbrumby, it's working for sure!

I meant the same function.

To clarify, before EventBridge pipes, one would have a lambda function on the dynamodb table stream which would send the dynamodb changes unmarshalled to EventBridge. And in an event-driven architecture all other listeners would only add rules/targets on EventBridge. And the bottleneck being a single lambda function on the dynamodb-stream.

Then AWS introduced EventBridge pipe which would be ideally the solution to replace this architecture, except the pipe simple transformer can't unmarshal the data, so you'd need to use a lambda function as transformer which would mean that you're still running the lambda function, you still have the dynamodb stream and you added a pipe in addition to that.