ApsaraMQ for Kafka:Create an AnalyticDB sink connector

Usage notes

• You can export data from a source topic of a Message Queue for Apache Kafka instance to AnalyticDB for MySQL or AnalyticDB for PostgreSQL through AnalyticDB for MySQL only if all services are in the same region. For more information about connector limits, see Limits.

• This feature is powered by Function Compute. Function Compute provides a free quota. Charges apply for resource usage that exceeds the free quota. For more information, see the Billing overview of Function Compute.

• You can query function call logs in Function Compute to troubleshoot issues. For more information, see Configure logging.

• During data export, Message Queue for Apache Kafka serializes messages as UTF-8 strings. Binary data formats are not supported.

• If the AnalyticDB sink connector uses a private endpoint, the Function Compute runtime environment cannot access it by default. To enable network connectivity, configure the function service in the Function Compute console with the same VPC and vSwitch as your destination AnalyticDB instance. For more information, see Update a service.

• Message Queue for Apache Kafka automatically creates a service-linked role when you create a connector.

  • If a service-linked role does not exist, Message Queue for Apache Kafka automatically creates one. This role grants Message Queue for Apache Kafka the permissions required to export data.

  • If a service-linked role already exists, Message Queue for Apache Kafka does not create another one.

  For more information about service-linked roles, see Service-linked roles.

Procedure

To export data from a source topic of a Message Queue for Apache Kafka instance to an AnalyticDB database with an AnalyticDB sink connector, follow these steps:

1. Optional: For the AnalyticDB Sink Connector, create the required topic and group

   If you do not need to customize the Topic and Group, you can skip this step and choose to automatically create them in the next step.

   Important

   Some topics required by the AnalyticDB sink connector must use local storage. For Message Queue for Apache Kafka instances running major version 0.10.2, you cannot manually create topics that use local storage. These topics can only be created automatically.

   1. Create the topics required by an AnalyticDB sink connector

   2. Create the group required by an AnalyticDB sink connector

2. Create and deploy an AnalyticDB sink connector

3. Configure services

   1. Configure the Function Compute service

   2. Configure the destination AnalyticDB database

4. Verify the result

   1. Send test messages

   2. Verify the result

Create required topics

You can manually create the five topics required by the AnalyticDB sink connector in the Message Queue for Apache Kafka console: a task offset topic, a task configuration topic, a task status topic, a dead-letter queue topic, and an error data topic. These topics have different requirements for the number of partitions and the storage engine. For more information, see the Parameters for configuring the source service table.

1. Log on to the ApsaraMQ for Kafka console.

2. On the Overview page, select a region in the Resource Distribution section.

   Important

   You must create topics in the same region as your application, which is the region where the ECS instance is deployed. Topics cannot be used across regions. For example, if a topic is created in the China (Beijing) region, the message producer and consumer must also run on an ECS instance in the China (Beijing) region.

3. On the Instances page, click the name of the target instance.

4. In the navigation pane on the left, click Topics.

5. On the Topics page, click Create Topic.

6. In the Create Topic panel, set the topic properties, and then click OK.

   Parameter	Description	Example
   Name	The name of the topic. Note In Kafka, topic names such as xxx_xxx and xxx.xxx are considered the same. An error is reported if you try to create a topic with a duplicate name.	demo
   Description	A brief description of the topic.	demo test
   Partitions	The number of partitions in the topic.	12
   Storage Engine Note Currently, you can select a storage engine type only for non-Serverless Professional Edition instances. For other instances, this parameter is not supported and is set to Cloud Storage by default.	The storage engine for topic messages. ApsaraMQ for Kafka supports the following two storage engines. Cloud Storage: The underlying layer uses Alibaba Cloud disks. This storage engine features low latency, high performance, high durability, and high reliability. It uses a distributed three-replica mechanism. If the Instance Edition of the instance is Standard (High Write), the storage engine can only be Cloud Storage. Local Storage: Uses the native Kafka in-sync replica (ISR) replication algorithm and a distributed three-replica mechanism.	Cloud Storage
   Message Type	The type of topic messages. Normal Message: By default, messages with the same key are distributed to the same partition, and messages within a partition are stored in the order they are sent. If a machine in the cluster goes down, messages may become disordered. When you set Storage Engine to Cloud Storage, Normal Message is selected by default. Partitionally Ordered Message: By default, messages with the same key are distributed to the same partition, and messages within a partition are stored in the order they are sent. If a machine in the cluster goes down, the order of messages is still guaranteed within the partition. However, sending messages to some partitions may fail. Normal operations resume after the partitions recover. When you set Storage Engine to Local Storage, Partitionally Ordered Message is selected by default.	Normal Message
   Log Cleanup Policy	The cleanup policy for topic logs. When you set Storage Engine to Local Storage, you must configure the Log Cleanup Policy. You can set Storage Engine to Local Storage only for Professional Edition instances. Standard Edition instances are not supported. ApsaraMQ for Kafka supports the following two cleanup policies. Delete: The default message cleanup policy. If disk capacity is sufficient, messages are retained for the maximum retention period. If disk capacity is insufficient (generally, when disk usage exceeds 85%), old messages are deleted early to ensure service availability. Compact: Uses the Kafka Log Compaction cleanup policy. The Log Compaction policy ensures that for messages with the same key, the latest value is always retained. This is mainly applicable to scenarios such as recovering state after a system breakdown or reloading the cache after a system restart. For example, when using Kafka Connect or Confluent Schema Registry, you must use a Kafka Compact Topic to store system state or configuration information. Important Compact topics are generally used only in certain ecosystem components, such as Kafka Connect or Confluent Schema Registry. Do not set this property for topics in other message sending and receiving scenarios. For more information, see ApsaraMQ for Kafka Demo Library.	Compact
   Tag	The tags of the topic.	demo

   After the topic is created, it is displayed in the list on the Topics page.

Create the group required by the AnalyticDB Sink Connector

You can manually create the group for the AnalyticDB Sink Connector data synchronization task in the Message Queue for Apache Kafka console. The name of the group must be connect-Task Name. For more information, see Source service parameters.

1. Log on to the ApsaraMQ for Kafka console.

2. On the Overview page, select a region in the Resource Distribution section.

3. On the Instances page, click the name of the target instance.

4. In the navigation pane on the left, click Groups.

5. On the Groups page, click Create Group.

6. In the Create Group panel, enter a name for the group in the Group ID text box, enter a brief description in the Description text box, add tags to the group, and then click OK.

   After the group is created, it is displayed in the list on the Groups page.

Create and deploy an AnalyticDB sink connector

1. Log on to the ApsaraMQ for Kafka console.

2. On the Overview page, select a region in the Resource Distribution section.

3. In the left-side navigation pane, click Connectors.

4. On the Connectors page, select the instance to which the connector belongs from the Select Instance drop-down list and click Create Connector.

5. In the Create Connector wizard, complete the following steps.

   1. On the Configure Basic Information tab, configure the parameters as needed, and then click Next.

      Parameter	Description	Example value
      Name	The name of the connector. Naming rules: The name can be 1 to 48 characters in length. The first character can be an uppercase letter, a lowercase letter, or a digit. The remaining characters can be lowercase letters, digits, or hyphens (-). The name must be unique within the same Message Queue for Apache Kafka instance. A data synchronization task for a Connector must use a Group named connect-task name. If you do not manually create the Group, the system will automatically create it for you.	kafka-adb-sink
      Instance	The instance name and ID are automatically populated.	demo alikafka_post-cn-st21p8vj****

   2. On the Configure Source Service tab, select Message Queue for Apache Kafka as the Data Source, configure the following parameters, and then click Next.

      Table 1. Source service parameters

      Parameter	Description	Example
      Data Source Topic	The topic from which to synchronize data.	adb-test-input
      Consumer Thread Concurrency	The number of concurrent consumer threads for the source topic. Default value: 6. Valid values: 1 2 3 6 12	6
      Consumer Offset	The position where consumption starts. Valid values: Earliest Offset: Starts consumption from the earliest available offset. Latest Offset: Starts consumption from the latest available offset.	Earliest Offset
      VPC ID	The VPC where the data synchronization task is located. This parameter is displayed after you click Configure Runtime Environment. By default, this is the VPC of the Message Queue for Apache Kafka instance and does not require manual entry.	vpc-bp1xpdnd3l***
      vSwitch ID	The vSwitch where the data synchronization task is located. This parameter is displayed after you click Configure Runtime Environment. This vSwitch must be in the same VPC as the Message Queue for Apache Kafka instance. By default, this is the vSwitch specified when you deployed the Message Queue for Apache Kafka instance.	vsw-bp1d2jgg81***
      Failure Handling Policy	The policy for handling partitions where a message sending failure occurs. This parameter is displayed after you click Configure Runtime Environment. Valid values: Continue Subscription: Continues subscribing to the partition and prints an error log. Stop Subscription: Stops subscribing to the partition and prints an error log. Note To learn how to view logs, see Connector-related operations. To find solutions based on error codes, see Error codes.	Continue Subscription
      Resource Creation Method	The method for creating the topics and group required by the connector. This parameter is displayed after you click Configure Runtime Environment. Auto Manual	Auto
      Connector Consumer Group	The group used by the data synchronization task of the connector. Click Configure Runtime Environment to display this parameter. The name of the group must be connect-task name.	connect-kafka-adb-sink
      Task Offset Topic	The topic used to store consumer offsets. This parameter is displayed after you click Configure Runtime Environment. Topic: It is recommended that the name starts with connect-offset. Partitions: The number of partitions must be greater than 1. Storage Engine: The topic must use local storage. cleanup.policy: The topic's log cleanup policy must be compact.	connect-offset-kafka-adb-sink
      Task Configuration Topic	The topic used to store task configurations. This parameter is displayed after you click Configure Runtime Environment. Topic: It is recommended that the name starts with connect-config. Partitions: The number of partitions must be 1. Storage Engine: The topic must use local storage. cleanup.policy: The topic's log cleanup policy must be compact.	connect-config-kafka-adb-sink
      Task Status Topic	The topic used to store task status. This parameter is displayed after you click Configure Runtime Environment. Topic: It is recommended that the name starts with connect-status. Partitions: The recommended number of partitions is 6. Storage Engine: The topic must use local storage. cleanup.policy: The topic's log cleanup policy must be compact.	connect-status-kafka-adb-sink
      Dead-letter Queue Topic	The topic that is used to store error data from the Connect framework. Click Configure Runtime Environment to display this parameter. This topic can be the same as the error data topic to save topic resources. Topic: It is recommended that the name starts with connect-error. Partitions: The recommended number of partitions is 6. Storage Engine: The topic can use local storage or cloud storage.	connect-error-kafka-adb-sink
      Error Data Topic	The topic used to store exception data from the Sink. Click Configure Runtime Environment to display this parameter. This topic can be the same as the dead-letter queue topic to save topic resources. Topic: It is recommended that the name starts with connect-error. Partitions: The recommended number of partitions is 6. Storage Engine: The topic can use local storage or cloud storage.	connect-error-kafka-adb-sink

   3. On the Configure Destination Service tab, select AnalyticDB as the Destination Service, configure the following parameters, and then click Create.

      Parameter	Description	Example
      Instance Type	The type of the destination AnalyticDB instance. Valid values: AnalyticDB for MySQL and AnalyticDB for PostgreSQL.	MySQL
      AnalyticDB Instance ID	The ID of the destination AnalyticDB for MySQL or AnalyticDB for PostgreSQL instance.	am-bp139yqk8u1ik****
      Database Name	The name of the destination database in the AnalyticDB instance.	adb_demo
      Table Name	The name of the destination table in the AnalyticDB database for storing messages.	user
      Database Username	The username for connecting to the AnalyticDB instance to import data.	adbmysql
      Database Password	The password for connecting to the AnalyticDB instance to import data. You set the password when you create the instance. If you forget the password, you can reset it. AnalyticDB for MySQL: For information about how to reset the password, see Change an account password. AnalyticDB for PostgreSQL: Log on to the AnalyticDB for PostgreSQL console, click the instance name, and then click Account Management in the navigation pane on the left. Find the account whose password you want to reset, and click Reset Password in the Actions column.	********

      Note

      Message Queue for Apache Kafka passes the username and password as environment variables to the Function Compute function when it creates the task. After the task is created, Message Queue for Apache Kafka does not save this information.

      After the connector is created, you can view it on the Connectors page.

6. On the Connectors page, find the connector that you created and click Deploy in the Actions column.

Configure the Function Compute service

After you successfully create and deploy an AnalyticDB Sink Connector in the Message Queue for Apache Kafka console, Function Compute automatically creates a service and a function for the connector. The service is named in the format kafka-service-<connector_name>-<random_string>, and the function is named in the format fc-adb-<random_string>.

1. On the Connectors page, find the target connector and click Configure Function in the Actions column.

   This action redirects you to the Function Compute console.

2. In the Function Compute console, find the automatically created function service and configure its VPC and vSwitch information. For specific steps, see Update a service.

Configure the destination AnalyticDB database

After you configure the Function Compute service, you must add the CIDR block of its vSwitch to the whitelist of your AnalyticDB instance. You can find this CIDR block on the vSwitch page of the VPC console, in the row corresponding to the VPC and vSwitch of the Function Compute service.

• AnalyticDB for MySQL: Log on to the AnalyticDB for MySQL console and configure the whitelist. For more information, see Configure an IP address whitelist.

• AnalyticDB for PostgreSQL: Log on to the AnalyticDB for PostgreSQL console and configure the whitelist. For more information, see Configure a whitelist.

Send test messages

1. On the Connectors page, find the target connector and click Test in the Actions column.

2. In the Send Message panel, send a test message.

   • Set Sending Method to Console.

     1. In the Message Key text box, enter the key of the message. For example, demo.

     2. In the Message Content text box, enter the content of the test message. For example, {"key": "test"}.

     3. Set Send to Specified Partition to specify whether to send the message to a specific partition.

        • Click Yes and enter the partition ID in the Partition ID text box. For example, 0. To query the partition ID, see View partition status.

        • Click No to not specify a partition.

   • Set Sending Method to Docker. Run the Docker command in the Run the Docker container to produce a sample message section to send the message.

   • Set Sending Method to SDK. Select an SDK for the required language or framework and a connection type to send messages.

Verify the result

After sending messages to the source topic in Message Queue for Apache Kafka, log on to the AnalyticDB for MySQL console or the AnalyticDB for PostgreSQL console, connect to the database, and go to the SQL Window of Data Management Service (DMS). Query the table in the corresponding instance to check if the data has been exported correctly.