# Using Lambda with Amazon MQ **Note** If you want to send data to a target other than a Lambda function or enrich the data before sending it, see [ Amazon EventBridge Pipes](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html). Amazon MQ is a managed message broker service for [Apache ActiveMQ](https://activemq.apache.org/) and [RabbitMQ](https://www.rabbitmq.com). A *message broker* enables software applications and components to communicate using various programming languages, operating systems, and formal messaging protocols through either topic or queue event destinations. Amazon MQ can also manage Amazon Elastic Compute Cloud (Amazon EC2) instances on your behalf by installing ActiveMQ or RabbitMQ brokers and by providing different network topologies and other infrastructure needs. You can use a Lambda function to process records from your Amazon MQ message broker. Lambda invokes your function through an [event source mapping](invocation-eventsourcemapping.md), a Lambda resource that reads messages from your broker and invokes the function [synchronously](invocation-sync.md). **Warning** Lambda event source mappings process each event at least once, and duplicate processing of records can occur. To avoid potential issues related to duplicate events, we strongly recommend that you make your function code idempotent. To learn more, see [How do I make my Lambda function idempotent](https://repost.aws/knowledge-center/lambda-function-idempotent) in the AWS Knowledge Center. The Amazon MQ event source mapping has the following configuration restrictions: + Concurrency – Lambda functions that use an Amazon MQ event source mapping have a default maximum [concurrency](lambda-concurrency.md) setting. For ActiveMQ, the Lambda service limits the number of concurrent execution environments to five per Amazon MQ event source mapping. For RabbitMQ, the number of concurrent execution environments is limited to 1 per Amazon MQ event source mapping. Even if you change your function's reserved or provisioned concurrency settings, the Lambda service won't make more execution environments available. To request an increase in the default maximum concurrency for a single Amazon MQ event source mapping, contact Support with the event source mapping UUID, as well as the region. Because increases are applied at the specific event source mapping level, not the account or region level, you need to manually request a scaling increase for each event source mapping. + Cross account – Lambda does not support cross-account processing. You cannot use Lambda to process records from an Amazon MQ message broker that is in a different AWS account. + Authentication – For ActiveMQ, only the ActiveMQ [SimpleAuthenticationPlugin](https://activemq.apache.org/security#simple-authentication-plugin) is supported. For RabbitMQ, only the [PLAIN](https://www.rabbitmq.com/access-control.html#mechanisms) authentication mechanism is supported. Users must use AWS Secrets Manager to manage their credentials. For more information about ActiveMQ authentication, see [Integrating ActiveMQ brokers with LDAP](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/security-authentication-authorization.html) in the *Amazon MQ Developer Guide*. + Connection quota – Brokers have a maximum number of allowed connections per wire-level protocol. This quota is based on the broker instance type. For more information, see the [Brokers](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-limits.html#broker-limits) section of **Quotas in Amazon MQ** in the *Amazon MQ Developer Guide*. + Connectivity – You can create brokers in a public or private virtual private cloud (VPC). For private VPCs, your Lambda function needs access to the VPC to receive messages. For more information, see [Configure network security](process-mq-messages-with-lambda.md#process-mq-messages-with-lambda-networkconfiguration) later in this section. + Event destinations – Only queue destinations are supported. However, you can use a virtual topic, which behaves as a topic internally while interacting with Lambda as a queue. For more information, see [Virtual Destinations](https://activemq.apache.org/virtual-destinations) on the Apache ActiveMQ website, and [Virtual Hosts](https://www.rabbitmq.com/vhosts.html) on the RabbitMQ website. + Network topology – For ActiveMQ, only one single-instance or standby broker is supported per event source mapping. For RabbitMQ, only one single-instance broker or cluster deployment is supported per event source mapping. Single-instance brokers require a failover endpoint. For more information about these broker deployment modes, see [Active MQ Broker Architecture](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-broker-architecture.html) and [Rabbit MQ Broker Architecture](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/rabbitmq-broker-architecture.html)in the *Amazon MQ Developer Guide*. + Protocols – Supported protocols depend on the type of Amazon MQ integration. + For ActiveMQ integrations, Lambda consumes messages using the OpenWire/Java Message Service (JMS) protocol. No other protocols are supported for consuming messages. Within the JMS protocol, only [https://activemq.apache.org/components/cms/api_docs/activemqcpp-3.6.0/html/classactivemq_1_1commands_1_1_active_m_q_text_message.html](https://activemq.apache.org/components/cms/api_docs/activemqcpp-3.6.0/html/classactivemq_1_1commands_1_1_active_m_q_text_message.html) and [https://activemq.apache.org/components/cms/api_docs/activemqcpp-3.9.0/html/classactivemq_1_1commands_1_1_active_m_q_bytes_message.html](https://activemq.apache.org/components/cms/api_docs/activemqcpp-3.9.0/html/classactivemq_1_1commands_1_1_active_m_q_bytes_message.html) are supported. Lambda also supports JMS custom properties. For more information about the OpenWire protocol, see [OpenWire](https://activemq.apache.org/openwire.html) on the Apache ActiveMQ website. + For RabbitMQ integrations, Lambda consumes messages using the AMQP 0-9-1 protocol. No other protocols are supported for consuming messages. For more information about RabbitMQ's implementation of the AMQP 0-9-1 protocol, see [AMQP 0-9-1 Complete Reference Guide](https://www.rabbitmq.com/amqp-0-9-1-reference.html) on the RabbitMQ website. Lambda automatically supports the latest versions of ActiveMQ and RabbitMQ that Amazon MQ supports. For the latest supported versions, see [Amazon MQ release notes](https://docs.aws.amazon.com/amazon-mq/latest/developer-guide/amazon-mq-release-notes.html) in the *Amazon MQ Developer Guide*. **Note** By default, Amazon MQ has a weekly maintenance window for brokers. During that window of time, brokers are unavailable. For brokers without standby, Lambda cannot process any messages during that window. **Topics** + [ ## Understanding the Lambda consumer group for Amazon MQ ](#services-mq-configure) + [ # Configuring Amazon MQ event source for Lambda ](process-mq-messages-with-lambda.md) + [ # Event source mapping parameters ](services-mq-params.md) + [ # Filter events from an Amazon MQ event source ](with-mq-filtering.md) + [ # Troubleshoot Amazon MQ event source mapping errors ](services-mq-errors.md) ## Understanding the Lambda consumer group for Amazon MQ To interact with Amazon MQ, Lambda creates a consumer group which can read from your Amazon MQ brokers. The consumer group is created with the same ID as the event source mapping UUID. For Amazon MQ event sources, Lambda batches records together and sends them to your function in a single payload. To control behavior, you can configure the batching window and batch size. Lambda pulls messages until it processes the payload size maximum of 6 MB, the batching window expires, or the number of records reaches the full batch size. For more information, see [Batching behavior](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching). The consumer group retrieves the messages as a BLOB of bytes, base64-encodes them into a single JSON payload, and then invokes your function. If your function returns an error for any of the messages in a batch, Lambda retries the whole batch of messages until processing succeeds or the messages expire. **Note** While Lambda functions typically have a maximum timeout limit of 15 minutes, event source mappings for Amazon MSK, self-managed Apache Kafka, Amazon DocumentDB, and Amazon MQ for ActiveMQ and RabbitMQ only support functions with maximum timeout limits of 14 minutes. This constraint ensures that the event source mapping can properly handle function errors and retries. You can monitor a given function's concurrency usage using the `ConcurrentExecutions` metric in Amazon CloudWatch. For more information about concurrency, see [Configuring reserved concurrency for a function](configuration-concurrency.md). **Example Amazon MQ record events** ``` { "eventSource": "aws:mq", "eventSourceArn": "arn:aws:mq:us-east-2:111122223333:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8", "messages": [ { "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-east-2.amazonaws.com.rproxy.govskope.us-37557-1234520418293-4:1:1:1:1", "messageType": "jms/text-message", "deliveryMode": 1, "replyTo": null, "type": null, "expiration": "60000", "priority": 1, "correlationId": "myJMSCoID", "redelivered": false, "destination": { "physicalName": "testQueue" }, "data":"QUJDOkFBQUE=", "timestamp": 1598827811958, "brokerInTime": 1598827811958, "brokerOutTime": 1598827811959, "properties": { "index": "1", "doAlarm": "false", "myCustomProperty": "value" } }, { "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-east-2.amazonaws.com.rproxy.govskope.us-37557-1234520418293-4:1:1:1:1", "messageType": "jms/bytes-message", "deliveryMode": 1, "replyTo": null, "type": null, "expiration": "60000", "priority": 2, "correlationId": "myJMSCoID1", "redelivered": false, "destination": { "physicalName": "testQueue" }, "data":"LQaGQ82S48k=", "timestamp": 1598827811958, "brokerInTime": 1598827811958, "brokerOutTime": 1598827811959, "properties": { "index": "1", "doAlarm": "false", "myCustomProperty": "value" } } ] } ``` ``` { "eventSource": "aws:rmq", "eventSourceArn": "arn:aws:mq:us-east-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8", "rmqMessagesByQueue": { "pizzaQueue::/": [ { "basicProperties": { "contentType": "text/plain", "contentEncoding": null, "headers": { "header1": { "bytes": [ 118, 97, 108, 117, 101, 49 ] }, "header2": { "bytes": [ 118, 97, 108, 117, 101, 50 ] }, "numberInHeader": 10 }, "deliveryMode": 1, "priority": 34, "correlationId": null, "replyTo": null, "expiration": "60000", "messageId": null, "timestamp": "Jan 1, 1970, 12:33:41 AM", "type": null, "userId": "AIDACKCEVSQ6C2EXAMPLE", "appId": null, "clusterId": null, "bodySize": 80 }, "redelivered": false, "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ==" } ] } } ``` In the RabbitMQ example, `pizzaQueue` is the name of the RabbitMQ queue, and `/` is the name of the virtual host. When receiving messages, the event source lists messages under `pizzaQueue::/`. # Configuring Amazon MQ event source for Lambda **Topics** + [ ## Configure network security ](#process-mq-messages-with-lambda-networkconfiguration) + [ ## Create the event source mapping ](#services-mq-eventsourcemapping) ## Configure network security To give Lambda full access to Amazon MQ through your event source mapping, either your broker must use a public endpoint (public IP address), or you must provide access to the Amazon VPC you created the broker in. When you use Amazon MQ with Lambda, create [AWS PrivateLink VPC endpoints](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html) that provide your function access to the resources in your Amazon VPC. **Note** AWS PrivateLink VPC endpoints are required for functions with event source mappings that use the default (on-demand) mode for event pollers. If your event source mapping uses [ provisioned mode](invocation-eventsourcemapping.md#invocation-eventsourcemapping-provisioned-mode), you don't need to configure AWS PrivateLink VPC endpoints. Create an endpoint to provide access to the following resources: + Lambda — Create an endpoint for the Lambda service principal. + AWS STS — Create an endpoint for the AWS STS in order for a service principal to assume a role on your behalf. + Secrets Manager — If your broker uses Secrets Manager to store credentials, create an endpoint for Secrets Manager. Alternatively, configure a NAT gateway on each public subnet in the Amazon VPC. For more information, see [Enable internet access for VPC-connected Lambda functions](configuration-vpc-internet.md). When you create an event source mapping for Amazon MQ, Lambda checks whether Elastic Network Interfaces (ENIs) are already present for the subnets and security groups configured for your Amazon VPC. If Lambda finds existing ENIs, it attempts to re-use them. Otherwise, Lambda creates new ENIs to connect to the event source and invoke your function. **Note** Lambda functions always run inside VPCs owned by the Lambda service. Your function's VPC configuration does not affect the event source mapping. Only the networking configuration of the event source's determines how Lambda connects to your event source. Configure the security groups for the Amazon VPC containing your broker. By default, Amazon MQ uses the following ports: `61617` (Amazon MQ for ActiveMQ), and `5671` (Amazon MQ for RabbitMQ). + Inbound rules – Allow all traffic on the default broker port for the security group associated with your event source. Alternatively, you can use a self-referencing security group rule to allow access from instances within the same security group. + Outbound rules – Allow all traffic on port `443` for external destinations if your function needs to communicate with AWS services. Alternatively, you can also use a self-referencing security group rule to limit access to the broker if you don't need to communicate with other AWS services. + Amazon VPC endpoint inbound rules — If you are using an Amazon VPC endpoint, the security group associated with your Amazon VPC endpoint must allow inbound traffic on port `443` from the broker security group. If your broker uses authentication, you can also restrict the endpoint policy for the Secrets Manager endpoint. To call the Secrets Manager API, Lambda uses your function role, not the Lambda service principal. **Example VPC endpoint policy — Secrets Manager endpoint** ``` { "Statement": [ { "Action": "secretsmanager:GetSecretValue", "Effect": "Allow", "Principal": { "AWS": [ "arn:aws::iam::123456789012:role/my-role" ] }, "Resource": "arn:aws::secretsmanager:us-west-2:123456789012:secret:my-secret" } ] } ``` When you use Amazon VPC endpoints, AWS routes your API calls to invoke your function using the endpoint's Elastic Network Interface (ENI). The Lambda service principal needs to call `lambda:InvokeFunction` on any roles and functions that use those ENIs. By default, Amazon VPC endpoints have open IAM policies that allow broad access to resources. Best practice is to restrict these policies to perform the needed actions using that endpoint. To ensure that your event source mapping is able to invoke your Lambda function, the VPC endpoint policy must allow the Lambda service principal to call `sts:AssumeRole` and `lambda:InvokeFunction`. Restricting your VPC endpoint policies to allow only API calls originating within your organization prevents the event source mapping from functioning properly, so `"Resource": "*"` is required in these policies. The following example VPC endpoint policies show how to grant the required access to the Lambda service principal for the AWS STS and Lambda endpoints. **Example VPC Endpoint policy — AWS STS endpoint** ``` { "Statement": [ { "Action": "sts:AssumeRole", "Effect": "Allow", "Principal": { "Service": [ "lambda.amazonaws.com" ] }, "Resource": "*" } ] } ``` **Example VPC Endpoint policy — Lambda endpoint** ``` { "Statement": [ { "Action": "lambda:InvokeFunction", "Effect": "Allow", "Principal": { "Service": [ "lambda.amazonaws.com" ] }, "Resource": "*" } ] } ``` ## Create the event source mapping Create an [event source mapping](invocation-eventsourcemapping.md) to tell Lambda to send records from an Amazon MQ broker to a Lambda function. You can create multiple event source mappings to process the same data with multiple functions, or to process items from multiple sources with a single function. To configure your function to read from Amazon MQ, add the required permissions and create an **MQ** trigger in the Lambda console. To read records from an Amazon MQ broker, your Lambda function needs the following permissions. You grant Lambda permission to interact with your Amazon MQ broker and its underlying resouces by adding permission statements to your function [execution role](lambda-intro-execution-role.md): + [mq:DescribeBroker](https://docs.aws.amazon.com/amazon-mq/latest/api-reference/brokers-broker-id.html#brokers-broker-id-http-methods) + [secretsmanager:GetSecretValue](https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html) + [ec2:CreateNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateNetworkInterface.html) + [ec2:DeleteNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DeleteNetworkInterface.html) + [ec2:DescribeNetworkInterfaces](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeNetworkInterfaces.html) + [ec2:DescribeSecurityGroups](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSecurityGroups.html) + [ec2:DescribeSubnets](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSubnets.html) + [ec2:DescribeVpcs](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeVpcs.html) + [logs:CreateLogGroup](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_CreateLogGroup.html) + [logs:CreateLogStream](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_CreateLogStream.html) + [logs:PutLogEvents](https://docs.aws.amazon.com/AmazonCloudWatchLogs/latest/APIReference/API_PutLogEvents.html) **Note** When using an encrypted customer managed key, add the `[kms:Decrypt](https://docs.aws.amazon.com/msk/1.0/apireference/clusters-clusterarn-bootstrap-brokers.html#clusters-clusterarn-bootstrap-brokersget)` permission as well. **To add permissions and create a trigger** 1. Open the [Functions page](https://console.aws.amazon.com/lambda/home#/functions) of the Lambda console. 1. Choose the name of a function. 1. Choose the **Configuration** tab, and then choose **Permissions**. 1. Under **Role name**, choose the link to your execution role. This link opens the role in the IAM console. ![\[\]](http://docs.aws.amazon.com/lambda/latest/dg/images/execution-role.png) 1. Choose **Add permissions**, and then choose **Create inline policy**. ![\[\]](http://docs.aws.amazon.com/lambda/latest/dg/images/inline-policy.png) 1. In the **Policy editor**, choose **JSON**. Enter the following policy. Your function needs these permissions to read from an Amazon MQ broker. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "mq:DescribeBroker", "secretsmanager:GetSecretValue", "ec2:CreateNetworkInterface", "ec2:DeleteNetworkInterface", "ec2:DescribeNetworkInterfaces", "ec2:DescribeSecurityGroups", "ec2:DescribeSubnets", "ec2:DescribeVpcs", "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "*" } ] } ``` ------ **Note** When using an encrypted customer managed key, you must also add the `kms:Decrypt` permission. 1. Choose **Next**. Enter a policy name and then choose **Create policy**. 1. Go back to your function in the Lambda console. Under **Function overview**, choose **Add trigger**. ![\[\]](http://docs.aws.amazon.com/lambda/latest/dg/images/add-trigger.png) 1. Choose the **MQ** trigger type. 1. Configure the required options, and then choose **Add**. Lambda supports the following options for Amazon MQ event sources: + **MQ broker** – Select an Amazon MQ broker. + **Batch size** – Set the maximum number of messages to retrieve in a single batch. + **Queue name** – Enter the Amazon MQ queue to consume. + **Source access configuration** – Enter virtual host information and the Secrets Manager secret that stores your broker credentials. + **Enable trigger** – Disable the trigger to stop processing records. To enable or disable the trigger (or delete it), choose the **MQ** trigger in the designer. To reconfigure the trigger, use the event source mapping API operations. # Event source mapping parameters All Lambda event source types share the same [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html) and [UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html) API operations. However, only some of the parameters apply to Amazon MQ and RabbitMQ. | Parameter | Required | Default | Notes | | --- | --- | --- | --- | | BatchSize | N | 100 | Maximum: 10,000 | | Enabled | N | true | none | | FunctionName | Y | N/A | none | | FilterCriteria | N | N/A | [Control which events Lambda sends to your function](invocation-eventfiltering.md) | | MaximumBatchingWindowInSeconds | N | 500 ms | [Batching behavior](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching) | | Queues | N | N/A | The name of the Amazon MQ broker destination queue to consume. | | SourceAccessConfigurations | N | N/A | For ActiveMQ, BASIC\$1AUTH credentials. For RabbitMQ, can contain both BASIC\$1AUTH credentials and VIRTUAL\$1HOST information. | # Filter events from an Amazon MQ event source You can use event filtering to control which records from a stream or queue Lambda sends to your function. For general information about how event filtering works, see [Control which events Lambda sends to your function](invocation-eventfiltering.md). This section focuses on event filtering for Amazon MQ event sources. **Note** Amazon MQ event source mappings only support filtering on the `data` key. **Topics** + [ ## Amazon MQ event filtering basics ](#filtering-AMQ) ## Amazon MQ event filtering basics Suppose your Amazon MQ message queue contains messages either in valid JSON format or as plain strings. An example record would look like the following, with the data converted to a Base64 encoded string in the `data` field. ------ #### [ ActiveMQ ] ``` { "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1---mq---us-east-2.amazonaws.com.rproxy.govskope.us-37557-1234520418293-4:1:1:1:1", "messageType": "jms/text-message", "deliveryMode": 1, "replyTo": null, "type": null, "expiration": "60000", "priority": 1, "correlationId": "myJMSCoID", "redelivered": false, "destination": { "physicalName": "testQueue" }, "data":"QUJDOkFBQUE=", "timestamp": 1598827811958, "brokerInTime": 1598827811958, "brokerOutTime": 1598827811959, "properties": { "index": "1", "doAlarm": "false", "myCustomProperty": "value" } } ``` ------ #### [ RabbitMQ ] ``` { "basicProperties": { "contentType": "text/plain", "contentEncoding": null, "headers": { "header1": { "bytes": [ 118, 97, 108, 117, 101, 49 ] }, "header2": { "bytes": [ 118, 97, 108, 117, 101, 50 ] }, "numberInHeader": 10 }, "deliveryMode": 1, "priority": 34, "correlationId": null, "replyTo": null, "expiration": "60000", "messageId": null, "timestamp": "Jan 1, 1970, 12:33:41 AM", "type": null, "userId": "AIDACKCEVSQ6C2EXAMPLE", "appId": null, "clusterId": null, "bodySize": 80 }, "redelivered": false, "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ==" } ``` ------ For both Active MQ and Rabbit MQ brokers, you can use event filtering to filter records using the `data` key. Suppose your Amazon MQ queue contains messages in the following JSON format. ``` { "timeout": 0, "IPAddress": "203.0.113.254" } ``` To filter only those records where the `timeout` field is greater than 0, the `FilterCriteria` object would be as follows. ``` { "Filters": [ { "Pattern": "{ \"data\" : { \"timeout\" : [ { \"numeric\": [ \">\", 0] } } ] } }" } ] } ``` For added clarity, here is the value of the filter's `Pattern` expanded in plain JSON. ``` { "data": { "timeout": [ { "numeric": [ ">", 0 ] } ] } } ``` You can add your filter using the console, AWS CLI or an AWS SAM template. ------ #### [ Console ] to add this filter using the console, follow the instructions in [Attaching filter criteria to an event source mapping (console)](invocation-eventfiltering.md#filtering-console) and enter the following string for the **Filter criteria**. ``` { "data" : { "timeout" : [ { "numeric": [ ">", 0 ] } ] } } ``` ------ #### [ AWS CLI ] To create a new event source mapping with these filter criteria using the AWS Command Line Interface (AWS CLI), run the following command. ``` aws lambda create-event-source-mapping \ --function-name my-function \ --event-source-arn arn:aws:mq:us-east-2:123456789012:broker:my-broker:b-8ac7cc01-5898-482d-be2f-a6b596050ea8 \ --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"timeout\" : [ { \"numeric\": [ \">\", 0 ] } ] } }"}]}' ``` To add these filter criteria to an existing event source mapping, run the following command. ``` aws lambda update-event-source-mapping \ --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \ --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"timeout\" : [ { \"numeric\": [ \">\", 0 ] } ] } }"}]}' ``` To add these filter criteria to an existing event source mapping, run the following command. ``` aws lambda update-event-source-mapping \ --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \ --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"timeout\" : [ { \"numeric\": [ \">\", 0 ] } ] } }"}]}' ``` ------ #### [ AWS SAM ] To add this filter using AWS SAM, add the following snippet to the YAML template for your event source. ``` FilterCriteria: Filters: - Pattern: '{ "data" : { "timeout" : [ { "numeric": [ ">", 0 ] } ] } }' ``` ------ With Amazon MQ, you can also filter records where the message is a plain string. Suppose you want to process only records where the message begins with "Result: ". The `FilterCriteria` object would look as follows. ``` { "Filters": [ { "Pattern": "{ \"data\" : [ { \"prefix\": \"Result: \" } ] }" } ] } ``` For added clarity, here is the value of the filter's `Pattern` expanded in plain JSON. ``` { "data": [ { "prefix": "Result: " } ] } ``` You can add your filter using the console, AWS CLI or an AWS SAM template. ------ #### [ Console ] To add this filter using the console, follow the instructions in [Attaching filter criteria to an event source mapping (console)](invocation-eventfiltering.md#filtering-console) and enter the following string for the **Filter criteria**. ``` { "data" : [ { "prefix": "Result: " } ] } ``` ------ #### [ AWS CLI ] To create a new event source mapping with these filter criteria using the AWS Command Line Interface (AWS CLI), run the following command. ``` aws lambda create-event-source-mapping \ --function-name my-function \ --event-source-arn arn:aws:mq:us-east-2:123456789012:broker:my-broker:b-8ac7cc01-5898-482d-be2f-a6b596050ea8 \ --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : [ { \"prefix\": \"Result: \" } ] }"}]}' ``` To add these filter criteria to an existing event source mapping, run the following command. ``` aws lambda update-event-source-mapping \ --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \ --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : [ { \"prefix\": \"Result: \" } ] }"}]}' ``` ------ #### [ AWS SAM ] To add this filter using AWS SAM, add the following snippet to the YAML template for your event source. ``` FilterCriteria: Filters: - Pattern: '{ "data" : [ { "prefix": "Result " } ] }' ``` ------ Amazon MQ messages must be UTF-8 encoded strings, either plain strings or in JSON format. That's because Lambda decodes Amazon MQ byte arrays into UTF-8 before applying filter criteria. If your messages use another encoding, such as UTF-16 or ASCII, or if the message format doesn't match the `FilterCriteria` format, Lambda processes metadata filters only. The following table summarizes the specific behavior: | Incoming message format | Filter pattern format for message properties | Resulting action | | --- | --- | --- | | Plain string | Plain string | Lambda filters based on your filter criteria. | | Plain string | No filter pattern for data properties | Lambda filters (on the other metadata properties only) based on your filter criteria. | | Plain string | Valid JSON | Lambda filters (on the other metadata properties only) based on your filter criteria. | | Valid JSON | Plain string | Lambda filters (on the other metadata properties only) based on your filter criteria. | | Valid JSON | No filter pattern for data properties | Lambda filters (on the other metadata properties only) based on your filter criteria. | | Valid JSON | Valid JSON | Lambda filters based on your filter criteria. | | Non-UTF-8 encoded string | JSON, plain string, or no pattern | Lambda filters (on the other metadata properties only) based on your filter criteria. | # Troubleshoot Amazon MQ event source mapping errors When a Lambda function encounters an unrecoverable error, your Amazon MQ consumer stops processing records. Any other consumers can continue processing, provided that they do not encounter the same error. To determine the potential cause of a stopped consumer, check the `StateTransitionReason` field in the return details of your `EventSourceMapping` for one of the following codes: **`ESM_CONFIG_NOT_VALID`** The event source mapping configuration is not valid. **`EVENT_SOURCE_AUTHN_ERROR`** Lambda failed to authenticate the event source. **`EVENT_SOURCE_AUTHZ_ERROR`** Lambda does not have the required permissions to access the event source. **`FUNCTION_CONFIG_NOT_VALID`** The function's configuration is not valid. Records also go unprocessed if Lambda drops them due to their size. The size limit for Lambda records is 6 MB. To redeliver messages upon function error, you can use a dead-letter queue (DLQ). For more information, see [Message Redelivery and DLQ Handling](https://activemq.apache.org/message-redelivery-and-dlq-handling) on the Apache ActiveMQ website and [Reliability Guide](https://www.rabbitmq.com/reliability.html) on the RabbitMQ website. **Note** Lambda does not support custom redelivery policies. Instead, Lambda uses a policy with the default values from the [Redelivery Policy](https://activemq.apache.org/redelivery-policy) page on the Apache ActiveMQ website, with `maximumRedeliveries` set to 6.