Todos os produtos
Search

Este produto

    ApsaraMQ for Kafka:Criar um conector sink do FC

    Central de documentação

    ApsaraMQ for Kafka:Criar um conector sink do FC

    Última atualização: Sep 21, 2026

    Crie um conector sink do Function Compute para exportar dados de um tópico de origem em sua instância do ApsaraMQ for Kafka para uma função do Function Compute.

    Pré-requisitos

    Antes de criar um conector sink do Function Compute, atenda aos seguintes pré-requisitos:

    • ApsaraMQ for Kafka

      • Ative o recurso de conector para a instância do ApsaraMQ for Kafka. Para mais informações, consulte Ative um conector.

      • Crie um tópico de origem para a instância do ApsaraMQ for Kafka. Para mais informações, consulte Etapa 1: Crie um tópico.

        Este exemplo utiliza um tópico chamado fc-test-input.

    • Function Compute

      • Crie uma função no Function Compute. Para mais informações, consulte Crie uma função.

        Importante

        A função deve ser uma função de evento.

        Este guia usa uma função de evento chamada hello_world como exemplo. A função está no serviço guide-hello_world e é executada em um ambiente de tempo de execução Python. O código de exemplo para a função é o seguinte:

        # -*- 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)

    • Opcional: EventBridge

      Nota

      Esta etapa é necessária apenas se a sua instância do ApsaraMQ for Kafka estiver na região China (Hangzhou) ou China (Chengdu).

    Observações

    • É possível exportar dados de um tópico de origem em uma instância do ApsaraMQ for Kafka para uma função no Function Compute apenas se ambos estiverem na mesma região. Para mais informações sobre as limitações do conector, consulte Limites.

    • Se a sua instância do ApsaraMQ for Kafka estiver na região China (Hangzhou) ou China (Chengdu), o conector será implantado no EventBridge.

      • O EventBridge é atualmente gratuito. Para mais informações, consulte Faturamento.

      • Ao criar um conector, o EventBridge cria automaticamente as seguintes funções vinculadas ao serviço: AliyunServiceRoleForEventBridgeSourceKafka e AliyunServiceRoleForEventBridgeConnectVPC.

        • Caso a função vinculada ao serviço ainda não exista, o EventBridge a criará automaticamente, permitindo que o EventBridge use essa função para acessar o ApsaraMQ for Kafka e a VPC.

        • Se essas funções vinculadas ao serviço já existirem, o EventBridge não as criará novamente.

        Para mais informações sobre funções vinculadas ao serviço, consulte funções vinculadas ao serviço.

      • Atualmente, não é possível visualizar os logs de tempo de execução das tarefas implantadas no EventBridge. Após a conclusão de uma tarefa do conector, verifique o progresso por meio do status de consumo do grupo de consumidores do tópico de origem. Para mais detalhes, consulte Visualize o status do consumidor.

    Procedimento

    Use um conector sink do Function Compute para exportar dados de um tópico de origem em uma instância do ApsaraMQ for Kafka para uma função no Function Compute:

    1. Opcional: Ative o acesso entre regiões para o conector sink do Function Compute

      Importante

      Se você não precisar de acesso entre regiões, pule esta etapa.

      Ative o acesso à internet para o conector sink do Function Compute

    2. Opcional: Ative o acesso entre contas para o conector sink do Function Compute

      Importante

      Se você não precisar de acesso entre contas, pule esta etapa.

    3. Opcional: Crie os tópicos e o grupo de consumidores necessários para o conector sink do Function Compute

      Importante

      • Caso não seja necessário personalizar os nomes dos tópicos e do grupo de consumidores, pule esta etapa.

      • Alguns tópicos exigidos por um conector sink do Function Compute devem usar o mecanismo de armazenamento local. Se a sua instância do ApsaraMQ for Kafka for da versão principal 0.10.2, não é possível criar manualmente tópicos com armazenamento local. Esses tópicos devem ser criados automaticamente.

      1. Crie os tópicos exigidos pelo conector sink do Function Compute

      2. Crie o grupo de consumidores exigido pelo conector sink do Function Compute

    4. Crie e implante um conector sink do Function Compute

    5. Verifique os resultados

      1. Envie uma mensagem de teste

      2. Visualize os logs da função

    Ativar o acesso à internet para um conector sink do FC

    Caso um conector sink do Function Compute precise acessar serviços da Alibaba Cloud em outras regiões, ative o acesso à internet para ele. Para mais detalhes, consulte Ative o acesso à internet para um conector.

    Criar uma política personalizada

    Na conta de destino, crie uma política personalizada para conceder acesso ao Function Compute.

    1. Faça login no console do RAM.

    2. No painel de navegação à esquerda, escolha Permissions > Policies.

    3. Na página Policies, clique em Create Policy.

    4. Na página Create Policy, configure a política.

      1. Na aba JSON, insira o script da política e clique em Next.

        O script de política a seguir concede permissões para acessar o Function Compute:

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

      2. Em Basic Information, insira KafkaConnectorFcAccess no campo Name.

      3. Clique em OK.

    Criar uma função do RAM

    Crie uma função do RAM na conta de destino. Não é possível selecionar o ApsaraMQ for Kafka como o serviço confiável ao criar uma função do RAM. Portanto, selecione outro serviço suportado e modifique manualmente a política de confiança após a criação da função.

    1. No painel de navegação à esquerda, escolha Identity Management > Roles.

    2. Na página Roles, clique em Create Role.

    3. No painel Create Role, configure a função.

      1. Selecione Alibaba Cloud Service como tipo de entidade confiável e clique em Next.

      2. Na seção Role Type, selecione Normal Service Role. Para Role Name, insira AliyunKafkaConnectorRole. Na lista suspensa Select Trusted Service, selecione Function Compute e clique em Complete.

    4. Na página Roles, localize e clique em AliyunKafkaConnectorRole.

    5. Na página de detalhes do AliyunKafkaConnectorRole, clique na aba Trust Policy Management e clique em Edit Trust Policy.

    6. No painel Edit Trust Policy, substitua fc no script por alikafka e clique em OK.

      Após salvar as alterações, na aba Trust Policy Management do AliyunKafkaConnectorRole, confirme se o parâmetro Service na política de confiança apresenta o valor alikafka.aliyuncs.com e se o parâmetro Action está definido como sts:AssumeRole.

    Adicionar permissões

    Na conta de destino, conceda permissões à função do RAM para acessar o Function Compute.

    1. No painel de navegação à esquerda, acesse Identity Management > Roles.

    2. Na página Roles, localize AliyunKafkaConnectorRole e clique em Add Permissions na coluna Actions.

    3. No painel Add Permissions, adicione a política KafkaConnectorFcAccess.

      1. Na seção Select Policy, selecione Custom Policy.

      2. Na lista Authorization Policy Name, localize e clique em KafkaConnectorFcAccess.

      3. Clique em OK.

      4. Clique em Complete.

    Criar tópicos para o conector sink do Function Compute

    No console do ApsaraMQ for Kafka, crie manualmente os cinco tópicos exigidos por um conector sink do Function Compute: um tópico de offset de tarefa, um tópico de configuração de tarefa, um tópico de status de tarefa, um tópico de fila de mensagens mortas e um tópico de dados de erro. Esses tópicos têm requisitos diferentes para o número de partições e o mecanismo de armazenamento. Para mais informações, consulte Parâmetros na etapa Configurar serviço de origem.

    1. Faça login no console do ApsaraMQ for Kafka.

    2. Na página Overview, selecione uma região na seção Resource Distribution.

      Importante

      Crie os tópicos na mesma região da sua aplicação, ou seja, a região de implantação da instância ECS. Não é possível usar tópicos entre regiões. Por exemplo, ao criar um tópico na região China (Beijing), o produtor e o consumidor de mensagens também devem ser executados em uma instância ECS na região China (Beijing).

    3. Na página Instances, clique no nome da instância de destino.

    4. No painel de navegação à esquerda, clique em Topics.

    5. Na página Topics, clique em Create Topic.

    6. No painel Create Topic, configure as definições do tópico e clique em OK.

      Parâmetro

      Descrição

      Exemplo

      Name

      O nome do tópico.

      Nota

      O Kafka considera nomes de tópicos com sublinhados (xxx_xxx) e pontos (xxx.xxx) idênticos. Caso tente criar um tópico duplicado, o sistema retornará um erro.

      demo

      Description

      Uma breve descrição do tópico.

      demo test

      Partitions

      O número de partições do tópico.

      12

      Storage Engine

      Nota

      Atualmente, a seleção do mecanismo de armazenamento está disponível apenas para instâncias não-serverless da Professional Edition. Para outros tipos de instância, esta opção não está disponível e o Cloud Storage é usado por padrão.

      O mecanismo de armazenamento para as mensagens no tópico.

      ApsaraMQ for Kafka suporta os dois seguintes mecanismos de armazenamento:

      • Cloud Storage: Este mecanismo usa discos da Alibaba Cloud para armazenamento de base e oferece alta performance, baixa latência e alta confiabilidade com um mecanismo distribuído de três réplicas. Se a Instance Edition for Standard (High Write), apenas o Cloud Storage estará disponível.

      • Local Storage: Este mecanismo usa o algoritmo nativo de replicação de réplicas em sincronia (ISR) do Kafka e um mecanismo distribuído de três réplicas.

      Cloud Storage

      Message Type

      O tipo de mensagens no tópico.

      • Normal Message: Por padrão, o Kafka distribui mensagens com a mesma chave para a mesma partição e as armazena na ordem de envio. Caso um nó do cluster falhe, as mensagens poderão ficar fora de ordem. Ao definir o Storage Engine como Cloud Storage, o sistema seleciona Normal Message por padrão.

      • Partitionally Ordered Message: Por padrão, o Kafka distribui mensagens com a mesma chave para a mesma partição e as armazena na ordem de envio. Mesmo que um nó do cluster falhe, a ordem das mensagens dentro da partição é garantida. No entanto, o envio de mensagens para algumas partições pode falhar até a recuperação delas. Ao definir o Storage Engine como Local Storage, o sistema seleciona Partitionally Ordered Message por padrão.

      Normal Message

      Log Cleanup Policy

      A política de limpeza de logs para o tópico.

      Ao selecionar Local Storage como Storage Engine (atualmente, apenas instâncias da Professional Edition suportam armazenamento local; esta opção não está disponível para a Standard Edition), configure a Log Cleanup Policy.

      ApsaraMQ for Kafka suporta as duas seguintes políticas de limpeza de logs.

      • Delete: A política padrão de limpeza de mensagens. Se o espaço em disco for suficiente, as mensagens são retidas dentro do período de retenção especificado. Se o espaço em disco for insuficiente (geralmente quando o uso do disco excede 85%), o sistema exclui as mensagens mais antigas antecipadamente para garantir a disponibilidade do serviço.

      • Compact: Usa a política de limpeza Kafka Log Compaction. A compactação de logs garante a retenção do valor mais recente para cada chave de mensagem. Isso é útil principalmente para cenários como restauração de estado após falha do sistema ou recarregamento de cache após reinicialização. Por exemplo, o Kafka Connect e o Confluent Schema Registry usam tópicos compactados para armazenar o estado do sistema e dados de configuração.

        Importante

        Tópicos compactados são normalmente usados apenas para componentes específicos do ecossistema, como Kafka Connect ou Confluent Schema Registry. Não defina esta propriedade para tópicos de produção e consumo geral de mensagens. Para mais informações, consulte a Biblioteca de Demos do ApsaraMQ for Kafka.

      Compact

      Tag

      As tags para o tópico.

      demo

      Após a criação, o tópico aparecerá na lista da página Topics.

    Criar grupo de consumidores para o conector sink do FC

    Crie manualmente o grupo de consumidores para uma tarefa de sincronização de dados do conector sink do Function Compute no console do ApsaraMQ for Kafka. O nome do grupo de consumidores deve ser connect-task-name. Para mais informações, consulte Parâmetros na etapa Configurar serviço de origem.

    1. Faça login no console do ApsaraMQ for Kafka.

    2. Na página Overview, selecione uma região na seção Resource Distribution.

    3. Na página Instances, clique no nome da instância de destino.

    4. No painel de navegação à esquerda, clique em Groups.

    5. Na página Groups, clique em Create Group.

    6. No painel Create Group, insira um nome para o grupo de consumidores na caixa de texto Group ID, insira uma breve descrição na caixa de texto Description, adicione tags ao grupo de consumidores e clique em OK.

      Após a criação, o grupo de consumidores aparecerá na lista da página Groups.

    Criar e implantar um conector sink do FC

    Crie e implante um conector sink do FC para sincronizar dados do ApsaraMQ for Kafka para o Function Compute.

    1. Faça login no console do ApsaraMQ for Kafka.

    2. Na página Overview, selecione uma região na seção Resource Distribution.

    3. No painel de navegação à esquerda, clique em Connectors.

    4. Na página Connectors, selecione a instância à qual o conector pertence na lista suspensa Select Instance e clique em Create Connector.

    5. No assistente Create Connector, conclua as seguintes etapas.

      1. Na aba Configure Basic Information, configure os seguintes parâmetros conforme necessário e clique em Next.

        Parâmetro

        Descrição

        Exemplo

        Name

        O nome do conector. O nome deve atender aos seguintes requisitos:

        • O nome pode ter até 48 caracteres e conter apenas dígitos, letras minúsculas e hifens (-). Não pode começar com um hífen (-).

        • O nome deve ser único dentro de uma instância do ApsaraMQ for Kafka.

        A tarefa de sincronização de dados do conector usa um grupo de consumidores chamado connect-task-name. Caso não crie este grupo de consumidores manualmente, o sistema o criará automaticamente.

        kafka-fc-sink

        Instance

        Por padrão, o nome e o ID da instância são exibidos.

        demo alikafka_post-cn-st21p8vj****

      2. Na aba Configure Source Service, defina Data Source como Message Queue for Apache Kafka, configure os seguintes parâmetros e clique em Next.

        Nota

        Caso já tenha criado um tópico e um grupo de consumidores, escolha a criação manual de recursos e insira as informações dos recursos existentes. Caso contrário, escolha a criação automática.

        Tabela 1. Parâmetros para configurar o serviço de origem

        Parâmetro

        Descrição

        Exemplo

        Data Source Topic

        O tópico de origem para sincronização de dados.

        fc-test-input

        Consumer Thread Concurrency

        O número de threads de consumo simultâneas para o tópico de origem. Valor padrão: 6. Valores válidos:

        • 1

        • 2

        • 3

        • 6

        • 12

        6

        Consumer Offset

        O offset a partir do qual iniciar o consumo. Valores válidos:

        • Earliest Offset: Inicia o consumo a partir do offset mais antigo.

        • Latest Offset: Inicia o consumo a partir do offset mais recente.

        Earliest Offset

        VPC ID

        O ID da VPC onde a tarefa de sincronização de dados é executada. Este parâmetro é exibido ao clicar em Configure Runtime Environment. Por padrão, corresponde à VPC da sua instância do ApsaraMQ for Kafka. Não é necessário especificar um valor.

        vpc-bp1xpdnd3l***

        vSwitch ID

        O ID do vSwitch onde a tarefa de sincronização de dados é executada. Este parâmetro é exibido ao clicar em Configure Runtime Environment. O vSwitch deve estar na mesma VPC da instância do ApsaraMQ for Kafka. Por padrão, corresponde ao vSwitch especificado durante a implantação da instância do ApsaraMQ for Kafka.

        vsw-bp1d2jgg81***

        Failure Handling Policy

        Define a ação quando a entrega de uma mensagem de uma partição falha. Este parâmetro é exibido ao clicar em Configure Runtime Environment. Valores válidos:

        • Continue Subscription: Continua o consumo da partição e registra o erro no log.

        • Stop Subscription: Interrompe o consumo da partição e registra o erro no log.

        Nota

        • Para mais informações sobre como visualizar logs, consulte Operações do conector.

        • Para mais informações sobre como encontrar soluções com base em códigos de erro, consulte Códigos de erro.

        Continue Subscription

        Resource Creation Method

        O método para criar o tópico e o grupo de consumidores exigidos pelo conector. Este parâmetro é exibido ao clicar em Configure Runtime Environment.

        • Auto

        • Manual

        Auto

        Connector Consumer Group

        O grupo de consumidores usado pela tarefa de sincronização de dados do conector. Este parâmetro é exibido ao clicar em Configure Runtime Environment. O nome do grupo de consumidores deve estar no formato connect-task-name.

        connect-kafka-fc-sink

        Task Offset Topic

        O tópico usado para armazenar offsets de consumo. Este parâmetro é exibido ao clicar em Configure Runtime Environment.

        • O nome do tópico deve começar com connect-offset.

        • Partições: O número de partições para o tópico deve ser maior que 1.

        • Mecanismo de armazenamento: O mecanismo de armazenamento para o tópico deve ser Local Storage.

        • cleanup.policy: A política de limpeza de logs para o tópico deve ser compact.

        connect-offset-kafka-fc-sink

        Task Configuration Topic

        O tópico usado para armazenar configurações de tarefas. Este parâmetro é exibido ao clicar em Configure Runtime Environment.

        • Tópico: Recomendamos que o nome do tópico comece com connect-config.

        • Partições: O número de partições para o tópico deve ser 1.

        • Mecanismo de armazenamento: O mecanismo de armazenamento para o tópico deve ser Local Storage.

        • cleanup.policy: A política de limpeza de logs para o tópico deve ser compact.

        connect-config-kafka-fc-sink

        Task Status Topic

        O tópico usado para armazenar o status das tarefas. Este parâmetro é exibido ao clicar em Configure Runtime Environment.

        • Tópico: Recomendamos que o nome do tópico comece com connect-status.

        • Partições: Recomendamos definir o número de partições como 6.

        • Mecanismo de armazenamento: O mecanismo de armazenamento para o tópico deve ser Local Storage.

        • cleanup.policy: A política de limpeza de logs para o tópico deve ser compact.

        connect-status-kafka-fc-sink

        Dead-letter Queue Topic

        O tópico usado para armazenar dados de erro do framework Connect. Este parâmetro é exibido ao clicar em Configure Runtime Environment. Para conservar recursos, use o mesmo tópico tanto para este parâmetro quanto para o parâmetro Error data topic.

        • Tópico: Recomendamos que o nome do tópico comece com connect-error.

        • Partições: Recomendamos definir o número de partições como 6.

        • Mecanismo de armazenamento: O mecanismo de armazenamento para o tópico pode ser Local Storage ou Cloud Storage.

        connect-error-kafka-fc-sink

        Error Data Topic

        O tópico usado para armazenar dados de erro do destino. Este parâmetro é exibido ao clicar em Configure Runtime Environment. Para economizar recursos, use o mesmo tópico para este parâmetro e para o parâmetro dead-letter queue topic.

        • Tópico: Recomendamos que o nome do tópico comece com connect-error.

        • Partições: Recomendamos definir o número de partições como 6.

        • Mecanismo de armazenamento: O mecanismo de armazenamento para o tópico pode ser Local Storage ou Cloud Storage.

        connect-error-kafka-fc-sink

      3. Na aba Configure Destination Service, selecione Function Compute para Target Service, configure os seguintes parâmetros e clique em Create.

        Nota

        Se a instância do conector estiver na região China (Hangzhou) ou China (Chengdu), o sistema exibirá a caixa de diálogo Service Authorization para a criação das funções vinculadas ao serviço AliyunServiceRoleForEventBridgeSourceKafka e AliyunServiceRoleForEventBridgeConnectVPC ao selecionar Function Compute como Target Service. Na caixa de diálogo Service Authorization, clique em Confirm, configure os seguintes parâmetros e clique em Create. Caso as funções vinculadas ao serviço já existam, a caixa de diálogo Service Authorization não será exibida.

        Parâmetro

        Descrição

        Exemplo

        Cross-account/Cross-region

        Define se o conector sink do FC sincroniza dados para um serviço do Function Compute em uma conta ou região diferente. Valor padrão: No. Valores válidos:

        • No: Modo mesma conta, mesma região.

        • Yes: Entre contas, entre regiões ou ambos.

        No

        Region

        A região do serviço do Function Compute. Por padrão, corresponde à região do conector sink do FC. Para sincronizar dados entre regiões, ative o acesso público do conector e selecione a região de destino. Para mais informações, consulte Ative o acesso público para um conector sink do FC.

        Importante

        Quando Cross-account/Cross-region estiver definido como Yes, o campo Region será exibido.

        cn-hangzhou

        Service Endpoint

        O endpoint do serviço do Function Compute. Obtenha o endpoint na seção Common Info, na página Overview do console do Function Compute.

        • Endpoint interno: Recomendado para baixa latência. Use este tipo de endpoint se a sua instância do ApsaraMQ for Kafka e o serviço do Function Compute estiverem na mesma região.

        • Endpoint público: Não recomendado devido à maior latência. Use este tipo de endpoint se a sua instância do ApsaraMQ for Kafka e o serviço do Function Compute estiverem em regiões diferentes. Para usar um endpoint público, ative o acesso público do conector. Para mais informações, consulte Ative o acesso público para um conector sink do FC.

        Importante

        Quando Cross-account/Cross-region estiver definido como Yes, o campo Service Endpoint será exibido.

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

        Alibaba Cloud Account

        O ID da conta da Alibaba Cloud à qual o serviço do Function Compute pertence. Obtenha o ID na seção Common Info, na página Overview do console do Function Compute.

        Importante

        Se Cross-account/Cross-region estiver definido como Yes, o campo Alibaba Cloud Account será exibido.

        188***

        RAM Role Name

        O nome da função do RAM assumida pelo ApsaraMQ for Kafka para acessar o serviço do Function Compute.

        Importante

        Quando Cross-account/Cross-region estiver definido como Yes, o campo RAM Role Name será exibido.

        AliyunKafkaConnectorRole

        Service Name

        O nome do serviço no Function Compute.

        guide-hello_world

        Function Name

        O nome da função no serviço do Function Compute.

        hello_world

        Version or Alias

        A versão ou alias do serviço do Function Compute.

        Importante

        • Se Cross-account/Cross-region estiver definido como No, selecione Specified Version ou Specified Alias.

        • Se Cross-account/Cross-region estiver definido como Yes, insira manualmente uma versão ou alias do serviço.

        LATEST

        Service Version

        A versão do serviço do Function Compute.

        Importante

        Se Cross-account/Cross-region estiver definido como No e Version or Alias estiver definido como Specified Version, o parâmetro Service Version será exibido.

        LATEST

        Service Alias

        O alias do serviço do Function Compute.

        Importante

        Quando Cross-account/Cross-region estiver definido como No e Version or Alias estiver definido como Specified Alias, o parâmetro Service Alias será exibido.

        jy

        Transmission Mode

        O modo para envio de mensagens. Valores válidos:

        • Asynchronous: Recomendado.

        • Synchronous: Não recomendado. Neste modo, o processamento lento de mensagens pelo Function Compute também desacelera o ApsaraMQ for Kafka. Caso o processamento de um lote de mensagens leve mais de 5 minutos, o sistema aciona um rebalanceamento de cliente no ApsaraMQ for Kafka.

        Asynchronous

        Data Size

        O número máximo de mensagens a serem incluídas em um único lote. O conector agrega mensagens em lotes que respeitam tanto este número quanto os limites de tamanho de requisição subjacentes (6 MB para síncrono, 128 KB para assíncrono). Por exemplo, se o modo de entrega for assíncrono, o tamanho do lote for 20 e você quiser enviar 18 mensagens, sendo 17 mensagens com tamanho total de 127 KB e uma mensagem com 200 KB, o conector agregará e enviará as 17 mensagens em um único lote. A mensagem restante, cujo tamanho excede 128 KB, será enviada em um lote separado.

        Nota

        Ao enviar uma mensagem, se a key for definida como nula, a requisição não a incluirá. Se o value for definido como nulo, a requisição não o incluirá.

        • Se o tamanho total das mensagens em um lote não exceder o limite de tamanho da requisição, a requisição conterá o conteúdo da mensagem. O código a seguir fornece um exemplo de requisição:

          [
    {
        "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
    }
]

        • Se o tamanho de uma única mensagem exceder o limite de tamanho da requisição, a requisição não incluirá o conteúdo da mensagem. O código a seguir fornece um exemplo de requisição:

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

          Para obter o conteúdo da mensagem, extraia-a com base em seu offset.

        50

        Retries

        O número de novas tentativas após a falha no envio de uma mensagem. O valor padrão é 2, com intervalo de 1 a 3. Alguns erros que causam falha no envio não suportam novas tentativas. O mapeamento entre os Códigos de erro e o suporte a novas tentativas é o seguinte:

        • 4XX: Não há suporte para novas tentativas em erros 4xx, exceto para 429.

        • 5XX: Há suporte para novas tentativas.

        Nota

        • O conector chama a operação InvokeFunction para enviar mensagens ao Function Compute.

        • Caso uma mensagem falhe após o número máximo de tentativas, o sistema a enviará para o tópico de fila de mensagens mortas. As mensagens nesse tópico não acionam as tarefas do conector do Function Compute novamente. Configure alertas para o tópico de fila de mensagens mortas para monitorar seu status em tempo real e tratar exceções rapidamente.

        2

        Após a criação, o conector estará disponível na página Connectors.

    6. Na página Connectors, localize o conector recém-criado e clique em Deploy na coluna Actions.

      Para configurar recursos do Function Compute, acesse More > Configure Function na coluna Actions. O sistema redirecionará você para o console do Function Compute para concluir a configuração.

    Enviar uma mensagem de teste

    Após implantar o conector sink do Function Compute, envie uma mensagem para o tópico de origem em sua instância do ApsaraMQ for Kafka para verificar a sincronização dos dados com o Function Compute.

    1. Na página Connectors, localize o conector de destino e clique em Test na coluna Actions.

    2. No painel Send Message, envie uma mensagem de teste.

      • Para Sending Method, selecione Console.

        1. No campo Message Key, insira uma chave de mensagem, como demo.

        2. No campo Message Content, insira o conteúdo da mensagem, como {"key": "test"}.

        3. Para Send to Specified Partition, escolha uma opção:

          • Clique em Yes e insira um ID de partição, como 0, no campo Partition ID. Para localizar o ID da partição, consulte Visualize o status da partição.

          • Clique em No para enviar a mensagem sem especificar uma partição.

      • Para Sending Method, selecione Docker e execute o comando da seção Run the Docker container to produce a sample message.

      • Para Sending Method, selecione SDK. Em seguida, escolha o SDK e o método de integração da sua linguagem ou framework preferido para enviar a mensagem.

    Logs da função

    Após enviar uma mensagem para o tópico de origem em sua instância do ApsaraMQ for Kafka, verifique os logs da função para confirmar que ela a recebeu. Para mais informações, consulte Configure os logs.

    Sua mensagem de teste aparecerá nos logs.

    Na página de detalhes da função, clique na aba Log query e selecione Advanced query. Selecione function-log como o Logstore. O campo message nos logs contém a mensagem do tópico do Kafka, por exemplo, [INFO] hello world:[{"key":"1","offset":1,"partition":0,"timestamp":1605598174308,"topic":"fc-test-input","value":"1"}]. Isso confirma que a função recebeu a mensagem com sucesso por meio do gatilho do Kafka.