ApsaraMQ for Kafka:Create an OSS sink connector

Prerequisites

Usage notes

• The connector exports data from a source topic in an ApsaraMQ for Kafka instance to Function Compute, and then from Function Compute to Object Storage Service (OSS). The ApsaraMQ for Kafka instance, Function Compute service, and OSS bucket must be in the same region. For more information about connector limitations, see Limitations.

• This feature is based on the Function Compute service. Function Compute provides a free quota. If you exceed the free quota, you are charged for any additional resources you use. For more information about billing, see Billing overview.

• Function Compute supports log queries for function invocations. For more information, see Configure logs.

• When exporting messages, ApsaraMQ for Kafka serializes them by using UTF-8 string encoding. Binary data formats are not supported.

Create and deploy an OSS sink connector

1. Log on to the ApsaraMQ for Kafka console.

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

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, perform the following steps:

   1. In the Configure Basic Information step, configure parameters and click Next. The following table describes the parameters.

      Important

      ApsaraMQ for Kafka automatically selects the Authorize to Create Service Linked Role option.

      • If a service-linked role has not been created, ApsaraMQ for Kafka automatically creates one for you. This role enables ApsaraMQ for Kafka to export data to OSS.

      • If a service-linked role already exists, ApsaraMQ for Kafka does not create a new one.

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

      Parameter	Description	Example
      Name	The name of the connector. The name must meet the following requirements: The name must be up to 48 characters in length and can contain only digits, lowercase letters, and hyphens (-). The name cannot start with a hyphen (-). The name must be unique within the same ApsaraMQ for 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-oss-sink
      Instance	By default, the name and ID of the instance are displayed.	demo alikafka_post-cn-st21p8vj****

   2. In the Configure Source Service step, select Message Queue for Apache Kafka as the source service, configure parameters, and then click Next. The following table describes the parameters.

      Parameter	Description	Example
      Data Source Topic	The topic from which to export data.	oss-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 offset from which consumption starts. Valid values: Earliest Offset: Consumption starts from the first available message in the partition. Latest Offset: Consumption starts from the next message produced after the connector starts.	Earliest Offset
      VPC ID	The VPC where the data synchronization task runs. Click Configure Runtime Environment to display this parameter. By default, this parameter is set to the VPC of the ApsaraMQ for Kafka instance and requires no configuration.	vpc-bp1xpdnd3l***
      vSwitch ID	The vSwitch for the data synchronization task. Click Configure Runtime Environment to display this parameter. The vSwitch must be in the same VPC as the ApsaraMQ for Kafka instance. By default, this is the vSwitch specified when the ApsaraMQ for Kafka instance was deployed.	vsw-bp1d2jgg81***
      Failure Handling Policy	Specifies the policy for handling a message delivery failure on a topic partition. Click Configure Runtime Environment to display this parameter. Valid values: Continue Subscription: Continues to subscribe to the topic partition where the error occurred and prints an error log. Stop Subscription: Stops subscribing to the topic partition where the error occurred and prints an error log. Note For information on how to view logs, see Connector-related operations. For information on how to find solutions based on error codes, see Error codes.	Continue Subscription
      Resource Creation Method	The creation method for the topics and consumer group required by the connector. Click Configure Runtime Environment to display this parameter. Auto Manual	Auto
      Connector Consumer Group	The consumer group used by the connector. Click Configure Runtime Environment to display this parameter. We recommend that the name of this consumer group starts with connect-cluster.	connect-cluster-kafka-oss-sink
      Task Offset Topic	The topic for storing consumer offsets. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-offset. Number of partitions: The number of partitions must be greater than 1. Storage engine: The storage engine of the topic must be Local storage. Note Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-offset-kafka-oss-sink
      Task Configuration Topic	The topic for storing task configurations. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-config. Number of partitions: The number of partitions must be 1. Storage engine: The storage engine of the topic must be Local storage. Note Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-config-kafka-oss-sink
      Task Status Topic	The topic for storing task status. Click Configure Runtime Environment to display this parameter. Topic: We recommend that the topic name starts with connect-status. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic must be Local storage. Note Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature. cleanup.policy: The log cleanup policy for the topic must be compact.	connect-status-kafka-oss-sink
      Dead-letter Queue Topic	The topic for storing exception data from the Kafka 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: We recommend that the topic name starts with connect-error. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic can be Local storage or Cloud storage. Note Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature.	connect-error-kafka-oss-sink
      Error Data Topic	The topic for storing 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: We recommend that the topic name starts with connect-error. Number of partitions: We recommend that you set the number of partitions to 6. Storage engine: The storage engine of the topic can be Local storage or Cloud storage. Note Only Professional Edition instances support selecting Local storage when you create a topic. Standard Edition instances do not support this feature.	connect-error-kafka-oss-sink

   3. In the Configure Destination Service step, select Object Storage Service as the destination service, set parameters, and then click Create. The following table describes the parameters.

      Parameter	Description	Example
      Bucket Name	The name of the OSS bucket.	bucket_test
      AccessKey ID	The AccessKey ID of your Alibaba Cloud account.	yourAccessKeyID
      AccessKey Secret	The AccessKey secret of your Alibaba Cloud account.	yourAccessKeySecret

      Make sure that your Alibaba Cloud account is granted the following permissions according to the principle of least privilege:

      {
          "Version": "1",
          "Statement": [
              {
                  "Action": [
                      "oss:GetObject",
                      "oss:PutObject"
                  ],
                  "Resource": "*",
                  "Effect": "Allow"
              }
          ]
      }

      Note

      ApsaraMQ for Kafka passes the AccessKey ID and AccessKey Secret to OSS as environment variables when it creates the task. After creation, ApsaraMQ for Kafka does not store these keys.

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

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

Send messages

You can send a message to the source topic in ApsaraMQ for Kafka to verify that data is exported to OSS.

1. On the Connectors page, find the connector that you want to manage and click Test in the Actions column.

2. In the Send Message panel, configure the parameters to send a message for testing.

   • If you set the Sending Method parameter to Console, perform the following steps:

     1. In the Message Key field, enter the message key. Example: demo.

     2. In the Message Content field, enter the message content. Example: {"key": "test"}.

     3. Configure the Send to Specified Partition parameter to specify whether to send the test message to a specific partition.

        • If you want to send the test message to a specific partition, click Yes and enter the partition ID in the Partition ID field. Example: 0. For information about how to query partition IDs, see View partition status.

        • If you do not want to send the test message to a specific partition, click No.

   • If you set the Sending Method parameter to Docker, run the Docker command in the Run the Docker container to produce a sample message section to send the test message.

   • If you set the Sending Method parameter to SDK, select an SDK for the required programming language or framework and an access method to send and subscribe to the test message.

Verify the results

After you send a message to the source topic in ApsaraMQ for Kafka, check the OSS file manager to verify that the data was exported. For more information, see File overview.

If new objects are generated in the OSS bucket, the data is synchronized to OSS.

[
    {
        "key":"123",
        "offset":4,
        "overflowFlag":true,
        "partition":0,
        "timestamp":1603779578478,
        "topic":"Test",
        "value":"1",
        "valueSize":272687
    }
]

Related operations

You can configure the Function Compute resources that are required by the OSS sink connector based on your requirements.