Create a RocketMQ sink connector

Create a sink connector

1. Log on to the ApsaraMQ for Kafka console. In the Resource Distribution section of the Overview page, select the region where the ApsaraMQ for Kafka instance that you want to manage resides.

2. In the left navigation pane, choose Connector Ecosystem Integration > Tasks.

3. On the Tasks page, click Create Task.

4. In the Create Task panel, set Task Name and Description, configure the following parameters, and then click Save.

   • Task creation

     1. In the Source step, select Message Queue for Apache Kafka as the Data Provider. Configure the following parameters, and then click Next.

        Parameter	Description	Example
        Region	The region of the source Message Queue for Apache Kafka instance.	China (Beijing)
        Kafka instance	The source Message Queue for Apache Kafka instance.	alikafka_post-cn-jte3****
        Topic	Select the topic for producing Message Queue for Apache Kafka messages.	demo-topic
        Group ID	The consumer group of the source instance. Quick Create: This is the recommended option. A group ID that is named in the GID_EVENTBRIDGE_xxx format is automatically created. Use Existing: Select an independent group ID. Do not use a group ID that is shared with other services to avoid disrupting existing message consumption.	Quick Create
        Consumer offset	The offset from which message consumption starts. Latest offset (latest) Earliest offset (earliest)	Latest offset (latest)
        Network configuration	The network type for routing messages. Basic Network Self-managed Internet	Basic Network
        VPC	Required only if Network configuration is set to Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Required only if Network configuration is set to Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        Security group	Required only if Network configuration is set to Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	The encoding format for the message content. For data sources that support binary transmission, select one of the following formats. We recommend Json if you do not have specific encoding requirements. Json: Encodes binary data as a JSON object into the payload using UTF-8. Text: Encodes binary data as a UTF-8 string into the payload. This is the default format. Binary: Encodes binary data as a Base64-encoded string into the payload.	Json
        Messages	Advanced configuration parameter. The maximum number of messages to send per batch. A request is sent only when the number of accumulated messages reaches the specified value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	Advanced configuration parameter. The interval for invoking a function. The system aggregates messages and sends them to the sink at each interval. Valid values: 0 to 15. A value of 0 specifies that messages are delivered immediately without a delay.	3

     2. In the Filtering step, set the Pattern Content to filter events. For more information, see Event patterns.

     3. In the Transformation step, configure data transformation to perform complex data processing, such as splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to cleanse message data.

     4. In the Sink step, select Message Queue for Apache RocketMQ as the Service Type and configure the following parameters.

        Parameter	Description	Example
        Version	The version of the Message Queue for Apache RocketMQ instance.	RocketMQ 4.x
        Instance ID	The destination Message Queue for Apache RocketMQ instance.	test
        Topic	The destination topic in the RocketMQ instance.	test
        Message Body	EventBridge extracts data from a message using a JSONPath expression and routes the specified content to the destination. Complete Data Data Extraction Fixed Value Template	Data Extraction $.data.body
        Custom Property	Select Template. You can define a template with variables. EventBridge extracts fields from the source data and uses the template to transform the data. Note To pass through all properties from the source message, use the configuration provided in the example.	Variables: { "userProperties":"$.data.userProperties", "msgId":"$.data.systemProperties.UNIQ_KEY" } Template: { "EB_SYS_EMBED_OBJECT":"${userProperties}", "UNIQ_KEY":"${msgId}" }
        Message Key	EventBridge extracts data from a message using a JSONPath expression and routes the specified content to the destination. Empty Data Extraction Fixed Value Template	Data Extraction $.data.systemProperties.KEYS
        Message Tag		Data Extraction $.data.systemProperties.TAGS

   • Task properties

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

5. Return to the Tasks page, locate the task you created, and click Enable in the Actions column.

6. In the Note dialog box, read the message and then click OK.

   The task takes 30 to 60 seconds to start after it is enabled. You can view the startup progress in the Status column on the Tasks page.

Other operations

On the Tasks page, find the target task and perform the following operations from the Actions column:

• Click Details to view the task's basic information, properties, and monitoring metrics on the Task page.

• Click Edit to modify the task configuration in the Edit Task panel.

• Click Enable or Pause, and then click OK in the Note dialog box.

• Click Delete, and then click OK in the Note dialog box.