Create an ApsaraMQ for RabbitMQ sink connector

Create a sink connector task

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-side navigation pane, choose Connector Ecosystem Integration > Tasks.

3. On the Tasks page, click Create Task.

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

   • Task creation

     1. In the Source step of the configuration wizard, set Data Provider to Message Queue for Apache Kafka, configure the following parameters, and then click Next.

        Parameter	Description	Example
        region	Select the region where the source Message Queue for Apache Kafka instance is located.	China (Beijing)
        Kafka instance	Select the source Message Queue for Apache Kafka instance that produces the messages.	alikafka_post-cn-jte3****
        topic	Select the topic that produces Message Queue for Apache Kafka messages.	demo-topic
        group ID	Select the group ID of the source instance. Quick Create: Recommended. The system automatically creates a group ID in the GID_EVENTBRIDGE_xxx format. Use Existing: Select a dedicated group ID. Do not share it with existing workloads. Otherwise, message publishing and subscription may be affected.	Quick Create
        consumer offset	Select the consumer offset from which messages are consumed. Latest offset (latest) Earliest offset (earliest)	Latest offset (latest)
        Network configuration	Select the network type that is used to route messages. Basic Network Self-managed Internet	Basic Network
        VPC	Select a VPC ID. This parameter is required if Network configuration is set to Self-managed Internet.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Select a vSwitch ID. This parameter is required if Network configuration is set to Self-managed Internet.	vsw-bp1gbjhj53hdjdkg****
        security group	Select a security group. This parameter is required if Network configuration is set to Self-managed Internet.	alikafka_pre-cn-7mz2****
        Data Format	The data format specifies the encoding method for the content of binary data from the source. Multiple data formats are supported. If you do not have specific encoding requirements, set the format to Json. Json (Encodes binary data into a UTF-8 JSON object within the payload.) Text (Encodes binary data into a UTF-8 string within the payload.) Binary (Base64-encodes binary data into a string within the payload.)	Json
        Messages	Advanced Settings parameter. The maximum number of messages per batch. A batch is sent when the number of accumulated messages reaches this value. Valid values: 1 to 10,000.	100
        Interval (Unit: Seconds)	Advanced Settings parameter. The maximum interval in seconds for sending a batch. Messages are sent at this interval even if the batch push count is not met. Valid values: 0 to 15. A value of 0 indicates immediate delivery.	3

     2. In the Filtering wizard, use the Pattern Content field to define an event pattern for filtering messages. For more information, see Event patterns.

     3. In the Transformation wizard, configure data cleansing to perform operations such as splitting, mapping, enriching, and dynamically routing data. For more information, see Use Function Compute to perform message data cleansing.

     4. In the Sink wizard, set Service Type to Message Queue for AMQP and configure the following parameters.

        Parameter	Description	Example
        Instance ID	Select the destination ApsaraMQ for RabbitMQ instance.	amqp-cn-zvp2pny6****
        vhost	Select an existing vhost.	test
        Destination Type	Exchange: A producer sends messages to an exchange, which then routes the messages to one or more queues. Queue: Each message is delivered to one or more queues.	Queue
        Exchange	The destination exchange in the ApsaraMQ for RabbitMQ instance. This parameter is required only when Destination Type is set to Exchange.	exchange
        Queue	The destination queue in the ApsaraMQ for RabbitMQ instance. This parameter is required only when Destination Type is set to Queue.	queue
        Message Routing Key	EventBridge uses a JSONPath expression to extract data from an event for the routing key. This parameter is required only when Destination Type is set to Exchange. Data Extraction Fixed Value	Data Extraction $.data.key
        Message Body	EventBridge uses a JSONPath expression to extract data from an event for the message body. Complete Data Data Extraction Fixed Value Template	Data Extraction $.data.body
        MessageId	EventBridge uses a JSONPath expression to extract data from an event for the message ID. Null Data Extraction Fixed Value Template	Data Extraction $.data.props.messageId
        Custom Property	EventBridge uses a JSONPath expression to extract data from an event for custom properties. Null Data Extraction Template	Data Extraction $.data.props

   • Task properties

     Configure the retry policy for delivery failures and specify how errors are handled. For more information, see retry policies and dead-letter queues.

5. Return to the Tasks page, find your task, and click Enable in the Actions column.

6. In the Note dialog box, review the information and click OK.

   After you enable the task, it may take 30 to 60 seconds to start. 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 in the Actions column.

• View task details: Click Details. On the Task page, you can view the basic information, properties, and monitoring metrics of the task.

• Edit the task: Click Edit. In the Edit Task panel, you can modify the task details and properties.

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

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