Use event streams to route messages between ApsaraMQ for RocketMQ instances - EventBridge

This topic describes how to use the event stream feature provided by EventBridge to route messages between ApsaraMQ for RocketMQ instances.

Prerequisites

EventBridge is activated and the required permissions are granted to a Resource Access Management (RAM) user. For more information, see Activate EventBridge and grant permissions to a RAM user.
ApsaraMQ for RocketMQ instances are purchased and deployed. Make sure that the instances are in the Running state. For more information, see Create an instance.

Background information

As a type of lightweight channel for processing end-to-end streaming data in real time, event streams allow you to filter and transform lightweight streaming data and synchronize data between data warehouses, between data processing programs, and between data analysis and data processing systems. You can use an event stream to route messages from a source ApsaraMQ for RocketMQ instance to a destination ApsaraMQ for RocketMQ instance without the need to define an event bus. For more information, see Overview.

Step 1: Create an event stream for the destination instance

Note

You must create an event stream for the destination instance. For example, if you want to route messages from an ApsaraMQ for RocketMQ instance that resides in the China (Beijing) region to an ApsaraMQ for RocketMQ instance that resides in the China (Hangzhou) region, you must create an event stream in the China (Hangzhou) region.

Log on to the EventBridge console.
In the top navigation bar, select a region.
In the left-side navigation pane, click Event Streams.
On the Event Streams page, click Create Event Stream.

On the Create Event Stream page, configure the Task Name and Description parameters and follow the on-screen instructions to configure other parameters. Then, click Save. The following section describes the parameters:

Task Creation

In the Source step, set the Data Provider parameter to ApsaraMQ for RocketMQ and configure other parameters. Then, click Next Step. The following table describes the parameters.

Parameter	Description	Example
Region	The region where the source ApsaraMQ for RocketMQ instance resides.	China (Hangzhou)
Version	The version of the source ApsaraMQ for RocketMQ instance.	RocketMQ 4.x
Instance	The source ApsaraMQ for RocketMQ instance in which the messages that you want to route are produced.	MQ_INST_115964845466****_ByBehioo
Topic	The topic on the source ApsaraMQ for RocketMQ instance in which the messages that you want to route are produced.	topic
Tag	The tag that is used to filter messages in the source ApsaraMQ for RocketMQ instance.	test
Group ID	The name of the consumer group on the source ApsaraMQ for RocketMQ instance. You must use a separate consumer group to create the event source. Do not use a consumer group that is in use. Otherwise, existing messages may fail to be sent and received.	GID_http_1
Consumer Offset	The offset from which messages are consumed.	Latest Offset
Data Format	The data format feature is used to encode binary data delivered from the source into a specific data format. In message routing scenarios, set this parameter to Binary.	Binary
Batch Push	The batch push feature helps you aggregate multiple events at a time. This feature is triggered if the condition that is specified by the Messages parameter or the Interval (Unit: Seconds) parameter is met. For example, if you set the Messages parameter to 100 and the Interval (Unit: Seconds) parameter to 15, the push is executed when the number of messages reaches 100 even if only 10 seconds are elapsed.	Enable
Messages	The maximum number of messages that can be sent in each function invocation. Requests are sent only when the number of messages in the backlog reaches the specified value. Valid values: 1 to 10000.	100
Interval (Unit: Seconds)	The time interval at which the function is invoked. The system sends the aggregated messages to Function Compute at the specified time interval. Valid values: 0 to 15. Unit: seconds. The value 0 specifies that messages are sent immediately after aggregation.	3

In the Filtering step, specify the rule used to filter events and click Next Step.
Note
Specific content in the message value cannot be used to filter events based on the configurations in the example. You do not need to configure the Transformation step in message routing scenarios.

In the Sink step, set the Service Type parameter to ApsaraMQ for RocketMQ and follow the on-screen instructions to configure other parameters. Then, click Save. The following table describes the parameters.

Parameter	Description	Example
Version	The version of the destination ApsaraMQ for RocketMQ instance.	RocketMQ 4.x
Instance ID	The ID of the destination ApsaraMQ for RocketMQ instance that you created.	test
Topic	The topic on the destination ApsaraMQ for RocketMQ instance that you created.	test
Message Body	EventBridge extracts specific data from events, decodes the data in the Base64 format, and then routes the data to the event target.	Binary Extraction `$.data.body`
Custom Property	Select Template. You can specify a template and define the variables that are required in the template. EventBridge extracts fields from an event and transforms the event based on the template. Note If you want to deliver all message attributes from the event source, we recommend that you use the sample code provided in the Example column of this table.	Variables: `{ "userProperties":"$.data.userProperties", "msgId":"$.data.systemProperties.UNIQ_KEY" }` Template: `{ "EB_SYS_EMBED_OBJECT":"${userProperties}", "UNIQ_KEY":"${msgId}" }`
Message Key	EventBridge extracts data from an event by using JSONPath and routes the specified content of the event to the event target.	Partial Event `$.data.systemProperties.KEYS`
Message Tag	EventBridge extracts data from an event by using JSONPath and routes the specified content of the event to the event target.	Partial Event `$.data.systemProperties.TAGS`

Task Property
Specify the retry policy and dead-letter queue for the event stream. For more information, see Retry policies and dead-letter queues.

Go back to the Event Streams page, find the event stream that you created, and then click Enable in the Actions column.
In the Note message, click OK.
The event stream requires 30 to 60 seconds to be enabled. You can view the progress in the Status column of the event stream on the Event Streams page.

Step 2: Test the event stream

Log on to the ApsaraMQ for RocketMQ console.
In the top navigation bar, select the region where the source instance that you specified in Step 1: Create an event stream for the destination instance resides.
In the left-side navigation pane, click Instances.
On the Instances page, find the source instance that you specified in Step 1: Create an event stream for the destination instance and click Details in the Actions column.
In the left-side navigation pane, click Topics.
On the page that appears, click the name of the topic that you specified in Step 1: Create an event stream for the destination instance.
In the upper-right corner of the Topic Details page, click Quick Start.
In the Start Message Production and Consumption panel, set the Sending Method parameter to Console, configure the Body, Message Key, and Message Tag parameters, and then click OK.
After the message is sent, the The message is sent. prompt and the ID of the message are displayed.
Go back to the Instances page.
On the Instances page, find the destination instance that you specified in Step 1: Create an event stream for the destination instance and click Details in the Actions column.
In the left-side navigation pane, click Topics.
On the page that appears, click the name of the topic that you specified in Step 1: Create an event stream for the destination instance.
On the Topic Details page, click the Message Query tab.
Set the Query Method and Time Range parameters and click Search.
Check whether the returned message ID, tag, and key are the same as those of the message that is produced.