Message Queue for RocketMQ

Procedure

1. Log on to the EventBridge console. In the left-side navigation pane, click Event Streams.

2. In the top navigation bar, select a region and then click Create Event Stream.

3. On the Create Event Stream page, set Task name and Description, configure the following parameters, and then click Save.

   • Task Creation

     1. In the Source step, set Data Provider to Message Queue for Apache RocketMQ, configure the following parameters, and then click Next Step.

        Parameter	Description	Example
        Region	Select the region where the source Message Queue for RocketMQ instance is located.	China (Hangzhou)
        Version	Select the version of the RocketMQ instance. The following versions are supported: RocketMQ 4.x RocketMQ 5.x	RocketMQ 5.x
        Instance	The source ApsaraMQ for RocketMQ instance.	rmq-cn-jte3w5i****
        Topic	The topic in the source Message Queue for RocketMQ instance.	topic
        Tag	Specify a tag to filter messages in the source instance.	test
        Group ID	The name of the RocketMQ consumer group. Quickly Create: Recommended. If you select this option, the system automatically creates a Group ID that starts with GID_EVENTBRIDGE_xxx. Use Existing API Destination: If you select this option, select a dedicated Group ID. To avoid business disruptions, do not reuse a Group ID that is already in use.	Quick Create
        Consumer Offset	Select the offset from which to start consuming messages. Latest offset: Start consuming messages from the latest offset. Earliest offset: Start consuming messages from the earliest offset. Specified timestamp: Start consuming messages from a specific point in time.	Specified timestamp
        Consumed At	The time to start consumption. This parameter is required only if you set consumer offset to Specified timestamp.	2025-03-10 17:27:01
        VPC	The VPC where the source instance is deployed. This parameter is automatically populated and cannot be changed.	vpc-****
        vSwitches	The vSwitch of the source instance. This parameter is automatically populated and cannot be changed.	vsw-****
        Security Group	The following options are supported. Quickly Create: Recommended. If you select this option, a security group named eb-<millisecond timestamp>-security_group is automatically created. Use Existing API Destination: If you select this option, you must select a security group in the VPC. We recommend that you use the default security group policy.	Quick Create
        Data format (body)	The encoding format for the message body. Json is the default. Json: Default. Encodes binary data in UTF-8 as a JSON object. Text: Encodes binary data in UTF-8 as a string. Binary: Encodes binary data in Base64 as a string.	Json
        Messages	The maximum number of messages to process in a single batch. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	The maximum time to wait before sending a batch. The system aggregates messages and sends them to the sink at this interval. Valid values: 0 to 15. A value of 0 means the batch is sent immediately.	3

     2. In the Filtering, Transformation, and Sink steps, configure event filtering rules, transformation rules, and the event sink. For more information about how to configure event transformations, see Use Function Compute to clean message data.

   • Task Attributes

     Configure the retry policy and dead-letter queue for the event stream. For more information, see Retries and dead-letter queues.

4. Return to the Event Streams page, find the event stream that you created, and click Enable in the Actions column.

   After you enable the event stream, the configuration may take 30 to 60 seconds to take effect. Check the startup progress in the Status column on the Event Streams page.

Sample event

{
    "specversion":"1.0",
    "id":"94ebc15f-f0db-4bbe-acce-56fb72fb****",
    "source":"acs:mq",
    "type":"mq:Topic:SendMessage",
    "subject":"acs:mq:cn-hangzhou:123456789098****:MQ_INST_123456789098****_BXhFHryi%TopicName",
    "datacontenttype":"application/json; charset=utf-8",
    "time":"2021-04-08T06:01:20.766Z",
    "aliyunpublishtime":"2021-04-08T06:01:20.725Z",
    "aliyuneventbusname":"BusName",
    "data":{
        "topic":"TopicName",
        "systemProperties":{
            "MIN_OFFSET":"0",
            "TRACE_ON":"true",
            "MAX_OFFSET":"8",
            "MSG_REGION":"cn-hangzhou",
            "KEYS":"systemProperties.KEYS",
            "CONSUME_START_TIME":1628577790396,
            "UNIQ_KEY":"AC14C305069E1B28CDFA3181CDA2****",
            "TAGS":"systemProperties.TAGS",
            "INSTANCE_ID":"MQ_INST_123456789098****_BXhFHryi"
        },
        "userProperties":{
        },
        "body":"TEST"
    }
}

See Event overview for an explanation of the parameters defined in the CloudEvents specification.

The parameters in the data field are described in the following table.