All Products
Search

This Product

    ApsaraMQ for Kafka:Create FC sink connector

    Document Center

    ApsaraMQ for Kafka:Create FC sink connector

    Last Updated:Jun 20, 2026

    Create a Function Compute sink connector to export data from a source topic in your ApsaraMQ for Kafka instance to a Function Compute function.

    Prerequisites

    Before you create a Function Compute sink connector, ensure the following prerequisites are met:

    • ApsaraMQ for Kafka

      • Enable the connector feature for the ApsaraMQ for Kafka instance. For more information, see Enable a connector.

      • Create a source topic for the ApsaraMQ for Kafka instance. For more information, see Step 1: Create a topic.

        This example uses a topic named fc-test-input.

    • Function Compute

      • Create a function in Function Compute. For more information, see Create a function.

        Important

        The function must be an event function.

        This guide uses an event function named hello_world as an example. The function is in the guide-hello_world service and runs in a Python runtime environment. The sample code for the function is as follows:

        # -*- coding: utf-8 -*-
import logging
# To enable the initializer feature
# Implement the initializer function as follows:
# def initializer(context):
#   logger = logging.getLogger()
#   logger.info('initializing')
def handler(event, context):
  logger = logging.getLogger()
  logger.info('hello world:' + bytes.decode(event))
  return 'hello world:' + bytes.decode(event)

    • Optional: EventBridge

      Note

      This step is required only if your ApsaraMQ for Kafka instance is in the China (Hangzhou) or China (Chengdu) region.

    Notes

    • You can export data from a source topic in an ApsaraMQ for Kafka instance to a function in Function Compute only if they are in the same region. For more information about connector limitations, see Limits.

    • If your ApsaraMQ for Kafka instance is in the China (Hangzhou) or China (Chengdu) region, the connector is deployed on EventBridge.

      • EventBridge is currently free of charge. For more information, see Billing.

      • When you create a connector, EventBridge automatically creates the following service-linked roles: AliyunServiceRoleForEventBridgeSourceKafka and AliyunServiceRoleForEventBridgeConnectVPC.

        • If a service-linked role has not been created, EventBridge will automatically create a corresponding service-linked role for you so that EventBridge can use this role to access ApsaraMQ for Kafka and VPC.

        • If these service-linked roles already exist, EventBridge does not create them again.

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

      • You cannot currently view the runtime logs of tasks deployed on EventBridge. After a connector task completes, check its progress by viewing the consumption status of the source topic's consumer group. For details, see View consumer status.

    Procedure

    Use a Function Compute sink connector to export data from a source topic in an ApsaraMQ for Kafka instance to a function in Function Compute:

    1. Optional: Enable cross-region access for the Function Compute sink connector

      Important

      If you do not need cross-region access, skip this step.

      Enable internet access for the Function Compute sink connector

    2. Optional: Enable cross-account access for the Function Compute sink connector

      Important

      If you do not need cross-account access, skip this step.

    3. Optional: Create the topics and consumer group required by the Function Compute sink connector

      Important

      • If you do not need to customize the names of the topics and consumer group, you can skip this step.

      • Some topics required by a Function Compute sink connector must use the local storage engine. If your ApsaraMQ for Kafka instance is major version 0.10.2, you cannot manually create topics with local storage. These topics must be created automatically.

      1. Create topics required by the Function Compute sink connector

      2. Create the consumer group required by the Function Compute sink connector

    4. Create and deploy a Function Compute sink connector

    5. Verify the results

      1. Send a test message

      2. View function logs

    Enable internet access for an FC Sink connector

    If a Function Compute sink connector needs to access Alibaba Cloud services in other regions, you need to enable internet access for it. For details, see Enable internet access for a connector.

    Create a custom policy

    In the target account, create a custom policy to grant access to Function Compute.

    1. Sign in to the RAM console.

    2. In the left-side navigation pane, choose Permissions > Policies.

    3. On the Policies page, click Create Policy.

    4. On the Create Policy page, configure the policy.

      1. On the JSON tab, enter the policy script and click Next.

        The following policy script grants permissions to access Function Compute:

        {
    "Version": "1",
    "Statement": [
        {
            "Action": [
                "fc:InvokeFunction",
                "fc:GetFunction"
            ],
            "Resource": "*",
            "Effect": "Allow"
        }
    ]
}

      2. Under Basic Information, enter KafkaConnectorFcAccess for Name.

      3. Click OK.

    Create a RAM role

    Create a RAM role in the target account. You cannot select ApsaraMQ for Kafka as the trusted service when creating a RAM role. Therefore, you must select another supported service instead, and then manually modify the trust policy after the role is created.

    1. In the left-side navigation pane, choose Identity Management > Roles.

    2. On the Roles page, click Create Role.

    3. In the Create Role panel, configure the role.

      1. Select Alibaba Cloud Service for the trusted entity type, and click Next.

      2. In the Role Type section, select Normal Service Role. For Role Name, enter AliyunKafkaConnectorRole. From the Select Trusted Service drop-down list, select Function Compute, and then click Complete.

    4. On the Roles page, find and click AliyunKafkaConnectorRole.

    5. On the AliyunKafkaConnectorRole details page, click the Trust Policy Management tab, and then click Edit Trust Policy.

    6. In the Edit Trust Policy panel, replace fc in the script with alikafka, and then click OK.

      After you save the changes, on the Trust Policy Management tab for AliyunKafkaConnectorRole, confirm that the Service parameter in the trust policy is updated to alikafka.aliyuncs.com and the Action parameter is set to sts:AssumeRole.

    Add permissions

    In the target account, grant the RAM role permissions to access Function Compute.

    1. In the navigation pane on the left, go to Identity Management > Roles.

    2. On the Roles page, find AliyunKafkaConnectorRole and click Add Permissions in the Actions column.

    3. In the Add Permissions panel, add the KafkaConnectorFcAccess policy.

      1. In the Select Policy section, select Custom Policy.

      2. In the Authorization Policy Name list, find and click KafkaConnectorFcAccess.

      3. Click OK.

      4. Click Complete.

    Create topics for Function Compute sink connector

    In the ApsaraMQ for Kafka console, you can manually create the five topics required by a Function Compute sink connector: 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 partition count and storage engine. For more information, see Parameters in the Configure Source Service step.

    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.

      Important

      You must create topics in the region where your Elastic Compute Service (ECS) instance is deployed. A topic cannot be used across regions. For example, if the producers and consumers of messages run on an ECS instance that is deployed in the China (Beijing) region, the topic must also be created in the China (Beijing) region.

    3. On the Instances page, click the name of the instance that you want to manage.

    4. In the left-side navigation pane, click Topics.

    5. On the Topics page, click Create Topic.

    6. In the Create Topic panel, configure the topic settings and click OK.

      Parameter

      Description

      Example

      Name

      The name of the topic.

      Note

      Kafka considers topic names with underscores (xxx_xxx) and periods (xxx.xxx) identical. If you try to create a duplicate topic, the system reports an error.

      demo

      Description

      A brief description of the topic.

      demo test

      Partitions

      The partition count of the topic.

      12

      Storage Engine

      Note

      Currently, you can select a storage engine only for non-serverless Professional Edition instances. For other instance types, this option is not available, and Cloud Storage is used by default.

      The storage engine for the messages in the topic.

      ApsaraMQ for Kafka supports the following two storage engines:

      • Cloud Storage: This engine uses Alibaba Cloud disks for underlying storage and provides high performance, low latency, and high reliability with a distributed, three-replica mechanism. If the Instance Edition of the instance is Standard (High Write), only Cloud Storage can be used.

      • Local Storage: This engine uses the native Kafka in-sync replica (ISR) replication algorithm and a distributed, three-replica mechanism.

      Cloud Storage

      Message Type

      The type of messages in the topic.

      • Normal Message: By default, Kafka distributes messages with the same key to the same partition and stores them in the order they are sent. If a node in the cluster fails, messages may be out of order. If you set the Storage Engine to Cloud Storage, the system selects Normal Message by default.

      • Partitionally Ordered Message: By default, Kafka distributes messages with the same key to the same partition and stores them in the order they are sent. Even if a cluster node fails, in-partition message order is guaranteed. However, sending messages to some partitions may fail until they recover. If you set the Storage Engine to Local Storage, the system selects Partitionally Ordered Message by default.

      Normal Message

      Log Cleanup Policy

      The log cleanup policy for the topic.

      When you select Local Storage as the Storage Engine (currently, only Professional Edition instances support local storage, and this option is not available for Standard Edition instances), you need to configure the Log Cleanup Policy.

      ApsaraMQ for Kafka supports the following two log cleanup policies.

      • Delete: The default message cleanup policy. If disk space is sufficient, messages are retained within the specified retention period. If disk space is insufficient (typically when disk usage exceeds 85%), the system deletes older messages early to ensure service availability.

      • Compact: Uses the Kafka Log Compaction cleanup policy. Log compaction ensures that the system retains the latest value for each message key. This is mainly used for scenarios such as restoring state after a system crash or reloading a cache after a system restart. For example, Kafka Connect and Confluent Schema Registry use compacted topics to store system state and configuration data.

        Important

        Compacted topics are typically used only for specific ecosystem components, such as Kafka Connect or Confluent Schema Registry. Do not set this property for topics used for general message production and consumption. For more information, see the ApsaraMQ for Kafka Demo Library.

      Compact

      Tag

      The tags for the topic.

      demo

      After you create the topic, it appears in the topic list on the Topics page.

    Create consumer group for the FC sink connector

    You can manually create the consumer group for a Function Compute sink connector data synchronization task in the ApsaraMQ for Kafka console. The name of the consumer group must be connect-task-name. For more information, see Parameters in the Configure Source Service step.

    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. On the Instances page, click the name of the instance that you want to manage.

    4. In the left-side navigation pane, click Groups.

    5. On the Groups page, click Create Group.

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

      Once created, the consumer group appears in the list on the Groups page.

    Create and deploy an FC Sink Connector

    Create and deploy an FC Sink Connector to synchronize data from ApsaraMQ for Kafka to Function Compute.

    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, complete the following steps.

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

        Parameter

        Description

        Example

        Name

        The name of the connector. The name must meet the following requirements:

        • The name can be up to 48 characters long and can contain only digits, lowercase letters, and hyphens (-). It cannot start with a hyphen (-).

        • The name must be unique within an ApsaraMQ for Kafka instance.

        The connector's data synchronization task uses a consumer group named connect-task-name. If you do not create this consumer group manually, the system creates it automatically.

        kafka-fc-sink

        Instance

        By default, the instance name and ID are displayed.

        demo alikafka_post-cn-st21p8vj****

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

        Note

        If you have already created a topic and a consumer group, choose manual resource creation and enter the information for your existing resources. Otherwise, choose automatic resource creation.

        Table 1. Parameters for configuring the source service

        Parameter

        Description

        Example

        Data Source Topic

        The source topic from which data is synchronized.

        fc-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 to start consumption. Valid values:

        • Earliest Offset: Start consumption from the earliest offset.

        • Latest Offset: Start consumption from the latest offset.

        Earliest Offset

        VPC ID

        The ID of the VPC where the data synchronization task runs. This parameter is displayed when you click Configure Runtime Environment. By default, this is the VPC of your ApsaraMQ for Kafka instance. You do not need to specify a value.

        vpc-bp1xpdnd3l***

        vSwitch ID

        The ID of the vSwitch where the data synchronization task runs. This parameter is displayed when you click Configure Runtime Environment. The vSwitch must be in the same VPC as the ApsaraMQ for Kafka instance. By default, this is the vSwitch you specified when you deployed the ApsaraMQ for Kafka instance.

        vsw-bp1d2jgg81***

        Failure Handling Policy

        Controls what happens when a message from a partition fails delivery. This parameter is displayed when you click Configure Runtime Environment. Valid values:

        • Continue Subscription: Continue to consume from the partition and log the error.

        • Stop Subscription: Stop consuming from the partition and log the error.

        Note

        • For more information about how to view logs, see Connector operations.

        • For more information about how to find solutions based on error codes, see Error codes.

        Continue Subscription

        Resource Creation Method

        The method used to create the topic and consumer group required by the connector. This parameter is displayed when you click Configure Runtime Environment.

        • Auto

        • Manual

        Auto

        Connector Consumer Group

        The consumer group used by the data synchronization task of the connector. This parameter is displayed when you click Configure Runtime Environment. The name of the consumer group must be in the connect-task-name format.

        connect-kafka-fc-sink

        Task Offset Topic

        The topic that is used to store consumer offsets. This parameter is displayed when you click Configure Runtime Environment.

        • The topic name should start with connect-offset.

        • Partitions: The number of partitions for the topic must be greater than 1.

        • Storage engine: The storage engine for the topic must be Local Storage.

        • cleanup.policy: The log cleanup policy for the topic must be compact.

        connect-offset-kafka-fc-sink

        Task Configuration Topic

        The topic that is used to store task configurations. This parameter is displayed when you click Configure Runtime Environment.

        • Topic: We recommend that the topic name start with connect-config.

        • Partitions: The number of partitions for the topic must be 1.

        • Storage engine: The storage engine for the topic must be Local Storage.

        • cleanup.policy: The log cleanup policy for the topic must be compact.

        connect-config-kafka-fc-sink

        Task Status Topic

        The topic that is used to store task status. This parameter is displayed when you click Configure Runtime Environment.

        • Topic: We recommend that the topic name start with connect-status.

        • Partitions: We recommend that you set the number of partitions to 6.

        • Storage engine: The storage engine for the topic must be Local Storage.

        • cleanup.policy: The log cleanup policy for the topic must be compact.

        connect-status-kafka-fc-sink

        Dead-letter Queue Topic

        The topic that is used to store error data from the Connect framework. This parameter is displayed when you click Configure Runtime Environment. To conserve topic resources, you can use the same topic for both this parameter and the Error data topic parameter.

        • Topic: We recommend that the topic name start with connect-error.

        • Partitions: We recommend that you set the number of partitions to 6.

        • Storage engine: The storage engine for the topic can be Local Storage or Cloud Storage.

        connect-error-kafka-fc-sink

        Error Data Topic

        The topic that is used to store error data from the sink. This parameter is displayed when you click Configure Runtime Environment. To save topic resources, you can use the same topic for this parameter and the dead-letter queue topic parameter.

        • Topic: We recommend that the topic name start with connect-error.

        • Partitions: We recommend that you set the number of partitions to 6.

        • Storage engine: The storage engine for the topic can be Local Storage or Cloud Storage.

        connect-error-kafka-fc-sink

      3. On the Configure Destination Service tab, select Function Compute for Target Service, configure the following parameters, and click Create.

        Note

        If the instance for the connector is in the China (Hangzhou) or China (Chengdu) region, a Service Authorization dialog box appears for creating the AliyunServiceRoleForEventBridgeSourceKafka and AliyunServiceRoleForEventBridgeConnectVPC service-linked roles when you select Function Compute as the Target Service. In the Service Authorization dialog box, click Confirm, and then configure the following parameters and click Create. If the service-linked roles have already been created, the Service Authorization dialog box does not appear.

        Parameter

        Description

        Example

        Cross-account/Cross-region

        Specifies whether the FC Sink Connector synchronizes data to a Function Compute service that is in a different account or region. Default value: No. Valid values:

        • No: Same-account, same-region mode.

        • Yes: Cross-account, cross-region, or both.

        No

        Region

        The region of the Function Compute service. By default, this is the region of the FC Sink Connector. To synchronize data across regions, you must enable public access for the connector and then select the destination region. For more information, see Enable public access for an FC Sink Connector.

        Important

        When Cross-account/Cross-region is set to Yes, Region is displayed.

        cn-hangzhou

        Service Endpoint

        The endpoint of the Function Compute service. You can obtain the endpoint from the Common Info section on the Overview page of the Function Compute console.

        • Internal endpoint: Recommended for low latency. Use this type of endpoint if your ApsaraMQ for Kafka instance and Function Compute service are in the same region.

        • Public endpoint: Not recommended due to higher latency. Use this type of endpoint if your ApsaraMQ for Kafka instance and Function Compute service are in different regions. To use a public endpoint, you must enable public access for the connector. For more information, see Enable public access for an FC Sink Connector.

        Important

        When Cross-account/Cross-region is set to Yes, the Service Endpoint is displayed.

        http://188***.cn-hangzhou.fc.aliyuncs.com

        Alibaba Cloud Account

        The ID of the Alibaba Cloud account to which the Function Compute service belongs. You can obtain the ID from the Common Info section on the Overview page of the Function Compute console.

        Important

        If Cross-account/Cross-region is set to Yes, Alibaba Cloud Account is displayed.

        188***

        RAM Role Name

        The name of the RAM role that ApsaraMQ for Kafka assumes to access the Function Compute service.

        Important

        When Cross-account/Cross-region is set to Yes, RAM Role Name is displayed.

        AliyunKafkaConnectorRole

        Service Name

        The name of the service in Function Compute.

        guide-hello_world

        Function Name

        The name of the function in the Function Compute service.

        hello_world

        Version or Alias

        The version or alias of the Function Compute service.

        Important

        • If Cross-account/Cross-region is set to No, you must select either Specified Version or Specified Alias.

        • If Cross-account/Cross-region is set to Yes, you must manually enter a service version or alias.

        LATEST

        Service Version

        The version of the Function Compute service.

        Important

        If Cross-account/Cross-region is set to No and Version or Alias is set to Specified Version, the Service Version parameter is displayed.

        LATEST

        Service Alias

        The alias of the Function Compute service.

        Important

        When Cross-account/Cross-region is set to No and Version or Alias is set to Specified Alias, the Service Alias parameter is displayed.

        jy

        Transmission Mode

        The mode for sending messages. Valid values:

        • Asynchronous: Recommended.

        • Synchronous: Not recommended. In this mode, slow message processing by Function Compute also slows down ApsaraMQ for Kafka. If a batch of messages takes more than 5 minutes to process, a client rebalance is triggered in ApsaraMQ for Kafka.

        Asynchronous

        Data Size

        The maximum number of messages to include in a single batch. The connector aggregates messages into batches that respect both this number and the underlying request size limits (6 MB for synchronous, 128 KB for asynchronous). For example, if the delivery mode is asynchronous, the batch size is 20, and you want to send 18 messages where 17 messages have a total size of 127 KB and one message has a size of 200 KB, the connector aggregates and sends the 17 messages in one batch. The connector sends the remaining message, whose size exceeds 128 KB, in a separate batch.

        Note

        If you set the key to null when you send a message, the request does not include the key. If you set the value to null, the request does not include the value.

        • If the total size of messages in a batch does not exceed the request size limit, the request contains the message content. The following code provides a sample request:

          [
    {
        "key":"this is the message's key2",
        "offset":8,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325438,
        "topic":"Test",
        "value":"this is the message's value2",
        "valueSize":28
    },
    {
        "key":"this is the message's key9",
        "offset":9,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325440,
        "topic":"Test",
        "value":"this is the message's value9",
        "valueSize":28
    },
    {
        "key":"this is the message's key12",
        "offset":10,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325442,
        "topic":"Test",
        "value":"this is the message's value12",
        "valueSize":29
    },
    {
        "key":"this is the message's key38",
        "offset":11,
        "overflowFlag":false,
        "partition":4,
        "timestamp":1603785325464,
        "topic":"Test",
        "value":"this is the message's value38",
        "valueSize":29
    }
]

        • If the size of a single message exceeds the request size limit, the request does not include the message content. The following code provides a sample request:

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

          To obtain the message content, you must pull the message based on its offset.

        50

        Retries

        The number of retries after a message fails to send. The default value is 2. The value range is 1 to 3. Some errors that cause message sending to fail do not support retries. The mapping between Error codes and retry support is as follows:

        • 4XX: Retries are not supported for any 4xx errors except for 429.

        • 5XX: Retries are supported.

        Note

        • The connector calls the InvokeFunction operation to send messages to Function Compute.

        • If a message fails to be sent after the maximum number of retries, it is sent to the dead-letter queue topic. Messages in a dead-letter queue topic do not trigger Function Compute connector tasks again. We recommend that you configure alerts for the dead-letter queue topic to monitor its status in real time and handle exceptions in a timely manner.

        2

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

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

      To configure Function Compute resources, choose More > Configure Function in the Actions column. You are redirected to the Function Compute console to complete the configuration.

    Send a test message

    After you deploy the Function Compute sink connector, you can send a message to the source topic in your ApsaraMQ for Kafka instance to verify that data is synchronized to Function Compute.

    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, send a test message.

      • For Sending Method, select Console.

        1. In the Message Key field, enter a message key, such as demo.

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

        3. For Send to Specified Partition, choose an option:

          • Click Yes and enter a partition ID, such as 0, in the Partition ID field. To find the partition ID, see View partition status.

          • Click No to send the message without specifying a partition.

      • For Sending Method, select Docker, and run the command from the Run the Docker container to produce a sample message section.

      • For Sending Method, select SDK. Then, choose the SDK and integration method for your preferred language or framework to send a message.

    Function logs

    After sending a message to the source topic in your ApsaraMQ for Kafka instance, check the function logs to verify that the function received it. For more information, see Configure logs.

    Your test message appears in the logs.

    On the function details page, click the Log query tab and select Advanced query. Select function-log as the Logstore. The message field in the logs contains the message from the Kafka topic, for example, [INFO] hello world:[{"key":"1","offset":1,"partition":0,"timestamp":1605598174308,"topic":"fc-test-input","value":"1"}]. This confirms that the function successfully received the message from the Kafka trigger.