ApsaraMQ for RocketMQ:Message queue for Apache Kafka

Create a message inflow task

1. Log on to the ApsaraMQ for RocketMQ console. In the left-side navigation pane, choose Connector Ecosystem Integration > Tasks.

2. In the top navigation bar, select a region, such as China (Hangzhou). On the Tasks page, click Create Task.

3. In the Create Task page, configure the Task Name and Description parameters. Then, follow the on-screen instructions to configure other parameters. The following section describes the parameters:

   • Task Creation

     1. In the Source step, 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 to import messages from.	MQ_INST_115964845466****_ByBeUp3p
        Topic	Select the topic to import messages from.	topic
        Group ID	Specify the consumer group ID of the source instance. To prevent interruptions to your existing services, use a dedicated group ID for the message inflow task instead of sharing one with your business applications.	GID_http_1
        Consumer offset	Select the position from which to start consuming messages. Latest offset: Consume only new messages generated after the task is created. Earliest offset: Consume all existing messages from the beginning of the topic.	Latest offset
        Network Configuration	Select the network type for routing messages. Basic network Internet	basic network
        VPC	Select the VPC ID. This parameter is required if the task runs in a VPC.	vpc-bp17fapfdj0dwzjkd****
        vSwitch	Select the vSwitch ID. This parameter is required if the task runs in a VPC.	vsw-bp1gbjhj53hdjdkg****
        Security Group	Select the security group ID. This parameter is required if the task runs in a VPC.	alikafka_pre-cn-7mz2****
        Data Format (Body)	The encoding format for binary data from the source. If you do not have specific encoding requirements, set this parameter to JSON. JSON: (Default) Encodes binary data into a JSON payload by using UTF-8. Text: Encodes binary data into a string payload by using UTF-8. Binary: Encodes binary data into a string payload by using Base64.	JSON
        Messages	The maximum number of messages to send in a single function invocation. A request is sent only when the number of accumulated messages reaches this value. Valid values: [1, 10000].	100
        Interval (Unit: Seconds)	The interval for invoking the function. The system aggregates messages and sends them to Function Compute at the specified interval. Valid values: [0, 15]. The unit is seconds. A value of 0 indicates that messages are delivered immediately without waiting.	3

     2. In the Filtering step, define a data pattern to filter data. For more information, see Message filtering.

     3. In the Transformation step, specify a data cleansing method to implement data processing capabilities such as splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to Clean Message Data.

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

        Parameter	Description	Example
        Version	The version of the ApsaraMQ for RocketMQ instance to which you want to route messages. Valid values: RocketMQ 4.x: ApsaraMQ for RocketMQ 4.x. RocketMQ 5.x: ApsaraMQ for RocketMQ 5.x.	RocketMQ 5.x
        Instance ID	The ApsaraMQ for RocketMQ instance to which you want to route messages.	rmq-cn-****
        Topic	The topic on the ApsaraMQ for RocketMQ instance to which you want to route messages.	topic
        Message Body	Complete Data Data Extraction Fixed Value Template	Data Extraction $.data.body
        Custom Property	Not Specified Data Extraction Template	Template Parameters: { "userProperties":"$.data.userProperties", "msgId":"$.data.systemProperties.UNIQ_KEY" } Template: { "EB_SYS_EMBED_OBJECT":"${userProperties}", "UNIQ_KEY":"${msgId}" }
        Message Key	Not Specified Data Extraction Fixed Value Template	Data Extraction $.data.systemProperties.KEYS
        Message Tag	Not Specified Data Extraction Fixed Value Template	Data Extraction $.data.systemProperties.TAGS

   • Task Properties

     Configure the retry policy that is used when events fail to be pushed and the method that is used to handle faults. For more information, see Retry policies and dead-letter queues.

4. Click Save. On the Tasks page, find the task that you created. When the status in the Status column changes from Starting to Running, the task is created.

More operations