ApsaraMQ for Kafka:Create an OSS Sink Connector (Legacy)

Step 1: Create a destination service resource

Create a bucket in the OSS console. For detailed steps, see Create a bucket in the console.

In this example, the bucket is named oss-sink-connector-bucket.

Step 2: Create and start an OSS sink connector

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 navigation pane on the left, choose Connector Ecosystem Integration > Sink.
3. On the Sink page, click Create Task.

4. In the Create Message Outflow Task panel, configure the following parameters and click OK.

   1. In the Basic Information section, set Task Name and set Message Outflow Task Type to Object Storage Service.

   2. In the Resource Settings section, configure the following parameters.

      Table 1. Source (ApsaraMQ for Kafka)

      Parameter	Description	Example
      Region	The region where the source Kafka instance resides.	China (Hangzhou)
      Kafka Instance	The ID of the Kafka instance that contains the data source.	alikafka_post-cn-9hdsbdhd****
      Topic	The topic in the Kafka instance that contains the data source.	guide-sink-topic
      Group ID	The group ID in the Kafka instance that contains the data source. Quick Creation: Automatically creates a group ID named GID_EVENTBRIDGE_xxx. Use Existing: Select an existing group. Select an independent group ID that is not used by other services. This prevents affecting existing message sending and receiving.	Use Existing
      Concurrent Quota (Number of Consumers)	The number of concurrent threads to consume data from the topic. The following list describes the relationship between threads and topic partitions: Number of topic partitions = Number of concurrent consumers: One thread consumes data from one topic partition. This is the recommended setting. Number of topic partitions > Number of concurrent consumers: Multiple concurrent consumers share the consumption of all partitions. Number of topic partitions < Number of concurrent consumers: One thread consumes data from one topic partition. The extra consumers are idle.	2
      Consumer Offset	Latest: Consumption starts from the latest offset. Earliest: Consumption starts from the earliest offset.	Latest
      Network Configuration	Select Self-managed Public Network to transfer data across borders. Otherwise, select Default Network.	Default Network

      Table 1. Destination (Object Storage Service)

      Parameter	Description	Example
      OSS Bucket	The OSS bucket you created.	oss-sink-connector-bucket
      Save Path	No Subdirectory Required: Data is saved to the path {Kafka Instance ID}/{Topic Name}. Time-based Partitioning: YYYY/MM/dd/HH: OSS file directory is {Kafka Instance ID}/{Topic Name}/YYYY/MM/dd/HH. YYYY/MM/dd: OSS file directory is {Kafka Instance ID}/{Topic Name}/YYYY/MM/dd. YYYYMMddHH: OSS file directory is {Kafka Instance ID}/{Topic Name}/YYYYMMddHH. YYYYMMdd: OSS file directory is {Kafka Instance ID}/{Topic Name}/YYYYMMdd. Note YYYY, MM, dd, and HH represent year, month, day, and hour, respectively.	alikafka_post-cn-9dhsaassdd****/guide-oss-sink-topic/YYYY/MM/dd/HH
      Advanced Configuration	When the accumulated messages meet either the Batch Aggregation File Size or the Batch Aggregation Event Window condition, new messages are written to a new file.	None
      Batch Aggregation File Size	The file size for aggregation. Valid values: [1,128]. Unit: MiB.	5
      Batch Aggregation Event Window	The time window for aggregation. Unit: minutes.	1

   After completing the configuration, go to the Message Outflow page. Find the OSS sink connector task you just created and click Start in the Actions column. When the Status changes from Starting to Running, the connector is successfully created.

Step 3: Test the OSS sink connector

1. On the Message Outflow page, click the source topic in the Event Source column of the OSS sink connector task.

2. On the topic details page, click Send Test Message.

3. In the Start to Send and Consume Message panel, configure the message content and click OK.

   Set Send Method to Console, set Message Key to oss-sink-k2, set Message Content to oss-sink-v2, and set Send to Specific Partition to No.

4. On the Message Outflow page, click the destination bucket in the Event Target column of the OSS sink connector task.

5. On the bucket page, in the navigation pane on the left, choose Object Management > Files and navigate to the deepest path in the bucket.

   If message files exported from Kafka (such as partition_3_of...) exist in the deepest path, the OSS sink connector has successfully written messages to the OSS bucket.

   You will see two types of objects in this path:

   • System metadata files: Named .oss_meta_file_partition_{partitionID}. The number of these files matches the number of partitions in the upstream topic. They record batch information and can be ignored.

   • Data files: Named partition_{partitionID}_offset_{offset}_{8-character random string}. If an object aggregates multiple messages from one partition, the offset in the object name is the smallest offset among those messages.

6. In the Actions column next to the relevant object, choose > Download.

7. Open the downloaded file and view the message content.

   {"topic":"guide-oss-sink-topic","partition":0,"offset":0,"timestamp":1681378474218,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
   {"topic":"guide-oss-sink-topic","partition":0,"offset":1,"timestamp":1681378491498,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}
   {"topic":"guide-oss-sink-topic","partition":0,"offset":2,"timestamp":1681378492515,"headers":{"headers":[],"isReadOnly":false},"key":"oss-sink-k2","value":"oss-sink-v2"}

   Messages are separated by line breaks.