Create a Kafka 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, set the Task Name and Description, configure the task as described below, and 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 step, define an event pattern in the Pattern Content field to filter requests. For more information, see Event patterns.

     3. In the Transformation step, configure data cleansing for complex processing like splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to cleanse message data.

     4. In the Sink step, set Service Type to ApsaraMQ for Kafka and configure the following parameters.

        Parameter	Description	Example
        Instance ID	Select the destination ApsaraMQ for Kafka instance.	test
        Topic	Select a topic in the destination instance.	test
        Acknowledgment Mode	Select the acknowledgement signal that ApsaraMQ for Kafka sends to the client after receiving data. None LeaderOnly All	None
        Message Value	EventBridge uses a JSONPath expression to extract specific content from an event and route it to the destination. Complete Data Data Extraction Fixed Value Template	Extract from Event $.data.value
        Message Key	EventBridge uses a JSONPath expression to extract specific content from an event and route it to the destination. Null Data Extraction Fixed Value Template	Extract from Event $.data.key

   • Task properties

     Configure the retry policy and dead-letter queue for this task. For more information, see Retry and dead-letter queues.

5. On the Tasks page, find your task 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 being enabled. You can view the startup progress in the Status column on the Tasks page.

Other operations

On the Tasks page, find the message outflow task that you want to manage and perform other operations in the Actions column. The following items describe the operations that you can perform:

• View the task details: Click Details in the Actions column. On the page that appears, view the basic information, properties, and monitoring metrics of the task.

• Change the task configurations: Click Edit in the Actions column. In the Edit Task panel, change the configurations of the task.

• Enable or disable the task: Click Enable or Pause in the Actions column. In the Note message, click OK.

• Delete the task: Click Delete in the Actions column. In the Note message, click OK.