Share via

Gatilho do Apache Kafka para Azure Functions

  • Artigo

Você pode usar o gatilho do Apache Kafka no Azure Functions para executar seu código de função em resposta a mensagens nos tópicos do Kafka. Também é possível usar uma associação de saída do Kafka para gravar de sua função em um tópico. Para obter informações sobre detalhes de instalação e configuração, consulte Visão geral sobre associações do Apache Kafka para Azure Functions.

Importante

As associações do Kafka só estão disponíveis para o Functions no plano Elastic Premium e plano Dedicado (Serviço de Aplicativo). São compatíveis apenas na versão 3.x e posterior do runtime do Functions.

Exemplo

O uso do gatilho depende da modalidade C# usada no aplicativo de funções, que pode ser um destes modos:

Uma função C# compilada da biblioteca de classes do processo de trabalho isolado é executada em um processo isolado do runtime.

Os atributos usados dependem do provedor de eventos específico.

O exemplo a seguir mostra uma função C# que lê e registra a mensagem do Kafka como um evento do Kafka:

[Function("KafkaTrigger")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(eventData)["Value"]}");
}

Para receber eventos em um lote, use uma matriz de cadeia de caracteres como entrada, conforme mostrado neste exemplo:

[Function("KafkaTriggerMany")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default",
                  IsBatched = true)] string[] events, FunctionContext context)
{
    foreach (var kevent in events)
    {
        var logger = context.GetLogger("KafkaFunction");
        logger.LogInformation($"C# Kafka trigger function processed a message: {JObject.Parse(kevent)["Value"]}");
    }

A seguinte função registra a mensagem e os cabeçalhos para o Evento do Kafka:

[Function("KafkaTriggerWithHeaders")]
public static void Run(
    [KafkaTrigger("BrokerList",
                  "topic",
                  Username = "ConfluentCloudUserName",
                  Password = "ConfluentCloudPassword",
                  Protocol = BrokerProtocol.SaslSsl,
                  AuthenticationMode = BrokerAuthenticationMode.Plain,
                  ConsumerGroup = "$Default")] string eventData, FunctionContext context)
{
    var eventJsonObject = JObject.Parse(eventData);
    var logger = context.GetLogger("KafkaFunction");
    logger.LogInformation($"C# Kafka trigger function processed a message: {eventJsonObject["Value"]}");
    var headersJArr = eventJsonObject["Headers"] as JArray;
    logger.LogInformation("Headers for this event: ");
    foreach (JObject header in headersJArr)
    {
        logger.LogInformation($"{header["Key"]} {System.Text.Encoding.UTF8.GetString((byte[])header["Value"])}");

    }
}

Para obter um conjunto completo de exemplos de .NET em funcionamento, consulte o repositório de extensão do Kafka.

Observação

Para obter um conjunto equivalente de exemplos de TypeScript, consulte o repositório de extensão do Kafka

As propriedades específicas do arquivo function.json dependem do provedor de eventos, que nestes exemplos são Confluent ou Hubs de Eventos do Azure. Os exemplos a seguir mostram um gatilho do Kafka para uma função que lê e registra uma mensagem do Kafka.

O seguinte function.json define o gatilho para o provedor específico:

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "event",
            "direction": "in",
            "topic": "topic",
            "brokerList": "%BrokerList%",
            "username": "%ConfluentCloudUserName%",
            "password": "%ConfluentCloudPassword%",
            "protocol": "saslSsl",
            "authenticationMode": "plain",
            "consumerGroup" : "$Default",
            "dataType": "string"
        }
    ]
}

O seguinte código é executado quando a função é disparada:

module.exports = async function (context, event) {
    // context.log.info(event)
    context.log.info(`JavaScript Kafka trigger function called for message ${event.Value}`);
};

Para receber eventos em um lote, defina o valor cardinality como many no arquivo function.json, conforme mostrado nestes exemplos:

{
    "bindings": [
        {
            "type": "kafkaTrigger",
            "name": "event",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%"
        }
    ]
}

O seguinte código analisa a matriz de eventos e registra os dados do evento:

module.exports = async function (context, events) {
    function print(event) {
        var eventJson = JSON.parse(event)
        context.log.info(`JavaScript Kafka trigger function called for message ${eventJson.Value}`);
    }
    events.map(print);
};

O seguinte código também registra os dados de cabeçalho:

module.exports = async function (context, event) {
  function print(kevent) {
    var keventJson = JSON.parse(kevent)
    context.log.info(`JavaScript Kafka trigger function called for message ${keventJson.Value}`);
    context.log.info(`Headers for this message:`)
    let headers =  keventJson.Headers;
    headers.forEach(element => {
        context.log.info(`Key: ${element.Key} Value:${Buffer.from(element.Value, 'base64')}`) 
    });
  }
  event.map(print);
};

É possível definir um esquema Avro genérico para o evento passado para o gatilho. O seguinte function.json define o gatilho para o provedor específico com um esquema Avro genérico:

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaAvroGenericSingle",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "authenticationMode" : "PLAIN",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

O seguinte código é executado quando a função é disparada:

module.exports = async function (context, event) {
    context.log.info(`JavaScript Kafka trigger function called for message ${JSON.stringify(event)}`);
};

Para obter um conjunto completo de exemplos de JavaScript em funcionamento, consulte o repositório de extensão do Kafka.

As propriedades específicas do arquivo function.json dependem do provedor de eventos, que nestes exemplos são Confluent ou Hubs de Eventos do Azure. Os exemplos a seguir mostram um gatilho do Kafka para uma função que lê e registra uma mensagem do Kafka.

O seguinte function.json define o gatilho para o provedor específico:

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvent",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%",
            "sslCaLocation": "confluent_cloud_cacert.pem"
        }
    ]
}

O seguinte código é executado quando a função é disparada:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

Write-Output "Powershell Kafka trigger function called for message $kafkaEvent.Value"

Para receber eventos em um lote, defina o valor cardinality como many no arquivo function.json, conforme mostrado nestes exemplos:

{
    "bindings": [
      {
            "type": "kafkaTrigger",
            "name": "kafkaEvent",
            "direction": "in",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "dataType" : "string",
            "topic" : "topic",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "brokerList" : "%BrokerList%",
            "sslCaLocation": "confluent_cloud_cacert.pem"
        }
    ]
}

O seguinte código analisa a matriz de eventos e registra os dados do evento:

using namespace System.Net

param($kafkaEvents, $TriggerMetadata)

$kafkaEvents
foreach ($kafkaEvent in $kafkaEvents) {
    $event = $kafkaEvent | ConvertFrom-Json -AsHashtable
    Write-Output "Powershell Kafka trigger function called for message $event.Value"
}

O seguinte código também registra os dados de cabeçalho:

using namespace System.Net

param($kafkaEvents, $TriggerMetadata)

foreach ($kafkaEvent in $kafkaEvents) {
    $kevent = $kafkaEvent | ConvertFrom-Json -AsHashtable
    Write-Output "Powershell Kafka trigger function called for message $kevent.Value"
    Write-Output "Headers for this message:"
    foreach ($header in $kevent.Headers) {
        $DecodedValue = [System.Text.Encoding]::Unicode.GetString([System.Convert]::FromBase64String($header.Value))
        $Key = $header.Key
        Write-Output "Key: $Key Value: $DecodedValue"
    }
}

É possível definir um esquema Avro genérico para o evento passado para o gatilho. O seguinte function.json define o gatilho para o provedor específico com um esquema Avro genérico:

{
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaEvent",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

O seguinte código é executado quando a função é disparada:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

Write-Output "Powershell Kafka trigger function called for message $kafkaEvent.Value"

Para obter um conjunto completo de exemplos de PowerShell em funcionamento, consulte o repositório de extensão do Kafka.

As propriedades específicas do arquivo function.json dependem do provedor de eventos, que nestes exemplos são Confluent ou Hubs de Eventos do Azure. Os exemplos a seguir mostram um gatilho do Kafka para uma função que lê e registra uma mensagem do Kafka.

O seguinte function.json define o gatilho para o provedor específico:

{
      "scriptFile": "main.py",
      "bindings": [
        {
          "type": "kafkaTrigger",
          "name": "kevent",
          "topic": "topic",
          "brokerList": "%BrokerList%",
          "username": "%ConfluentCloudUserName%",
          "password": "%ConfluentCloudPassword%",
          "consumerGroup" : "functions",
          "protocol": "saslSsl",
          "authenticationMode": "plain"
        }
    ]
}

O seguinte código é executado quando a função é disparada:

import logging
from azure.functions import KafkaEvent

def main(kevent : KafkaEvent):
    logging.info(kevent.get_body().decode('utf-8'))
    logging.info(kevent.metadata)

Para receber eventos em um lote, defina o valor cardinality como many no arquivo function.json, conforme mostrado nestes exemplos:

{
      "scriptFile": "main.py",
      "bindings": [
        {
            "type" : "kafkaTrigger",
            "direction": "in",
            "name" : "kevents",
            "protocol" : "SASLSSL",
            "password" : "%ConfluentCloudPassword%",
            "topic" : "message_python",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "dataType": "string",
            "consumerGroup" : "$Default",
            "username" : "%ConfluentCloudUserName%",
            "BrokerList" : "%BrokerList%"    
        }
    ]
}

O seguinte código analisa a matriz de eventos e registra os dados do evento:

import logging
import typing
from azure.functions import KafkaEvent

def main(kevents : typing.List[KafkaEvent]):
    for event in kevents:
        logging.info(event.get_body())

O seguinte código também registra os dados de cabeçalho:

import logging
import typing
from azure.functions import KafkaEvent
import json
import base64

def main(kevents : typing.List[KafkaEvent]):
    for event in kevents:
        event_dec = event.get_body().decode('utf-8')
        event_json = json.loads(event_dec)
        logging.info("Python Kafka trigger function called for message " + event_json["Value"])
        headers = event_json["Headers"]
        for header in headers:
            logging.info("Key: "+ header['Key'] + " Value: "+ str(base64.b64decode(header['Value']).decode('ascii')))

É possível definir um esquema Avro genérico para o evento passado para o gatilho. O seguinte function.json define o gatilho para o provedor específico com um esquema Avro genérico:

{
  "scriptFile": "main.py",
  "bindings" : [ {
    "type" : "kafkaTrigger",
    "direction" : "in",
    "name" : "kafkaTriggerAvroGeneric",
    "protocol" : "SASLSSL",
    "password" : "ConfluentCloudPassword",
    "topic" : "topic",
    "authenticationMode" : "PLAIN",
    "avroSchema" : "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}",
    "consumerGroup" : "$Default",
    "username" : "ConfluentCloudUsername",
    "brokerList" : "%BrokerList%"
  } ]
}

O seguinte código é executado quando a função é disparada:

import logging
from azure.functions import KafkaEvent

def main(kafkaTriggerAvroGeneric : KafkaEvent):
    logging.info(kafkaTriggerAvroGeneric.get_body().decode('utf-8'))
    logging.info(kafkaTriggerAvroGeneric.metadata)

Para obter um conjunto completo de exemplos de Python em funcionamento, consulte o repositório de extensão do Kafka.

As anotações que você usa para configurar o gatilho dependem do provedor de eventos específico.

O exemplo a seguir mostra uma função Java que lê e registra o conteúdo do evento do Kafka:

@FunctionName("KafkaTrigger")
public void runSingle(
        @KafkaTrigger(
            name = "KafkaTrigger",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            dataType = "string"
         ) String kafkaEventData,
        final ExecutionContext context) {
        context.getLogger().info(kafkaEventData);
}

Para receber eventos em um lote, use uma cadeia de caracteres de entrada como uma matriz, conforme mostrado neste exemplo:

@FunctionName("KafkaTriggerMany")
public void runMany(
        @KafkaTrigger(
            name = "kafkaTriggerMany",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            cardinality = Cardinality.MANY,
            dataType = "string"
         ) String[] kafkaEvents,
        final ExecutionContext context) {
        for (String kevent: kafkaEvents) {
            context.getLogger().info(kevent);
        }    
}

A seguinte função registra a mensagem e os cabeçalhos para o Evento do Kafka:

@FunctionName("KafkaTriggerManyWithHeaders")
public void runSingle(
        @KafkaTrigger(
            name = "KafkaTrigger",
            topic = "topic",  
            brokerList="%BrokerList%",
            consumerGroup="$Default", 
            username = "%ConfluentCloudUsername%", 
            password = "ConfluentCloudPassword",
            authenticationMode = BrokerAuthenticationMode.PLAIN,
            protocol = BrokerProtocol.SASLSSL,
            // sslCaLocation = "confluent_cloud_cacert.pem", // Enable this line for windows.
            dataType = "string",
            cardinality = Cardinality.MANY
         ) List<String> kafkaEvents,
        final ExecutionContext context) {
            Gson gson = new Gson(); 
            for (String keventstr: kafkaEvents) {
                KafkaEntity kevent = gson.fromJson(keventstr, KafkaEntity.class);
                context.getLogger().info("Java Kafka trigger function called for message: " + kevent.Value);
                context.getLogger().info("Headers for the message:");
                for (KafkaHeaders header : kevent.Headers) {
                    String decodedValue = new String(Base64.getDecoder().decode(header.Value));
                    context.getLogger().info("Key:" + header.Key + " Value:" + decodedValue);                    
                }                
            }
        }

É possível definir um esquema Avro genérico para o evento passado para o gatilho. A seguinte função define o gatilho para o provedor específico com um esquema Avro genérico:

private static final String schema = "{\"type\":\"record\",\"name\":\"Payment\",\"namespace\":\"io.confluent.examples.clients.basicavro\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"type\",\"type\":\"string\"}]}";

@FunctionName("KafkaAvroGenericTrigger")
public void runOne(
        @KafkaTrigger(
                name = "kafkaAvroGenericSingle",
                topic = "topic",
                brokerList="%BrokerList%",
                consumerGroup="$Default",
                username = "ConfluentCloudUsername",
                password = "ConfluentCloudPassword",
                avroSchema = schema,
                authenticationMode = BrokerAuthenticationMode.PLAIN,
                protocol = BrokerProtocol.SASLSSL) Payment payment,
        final ExecutionContext context) {
    context.getLogger().info(payment.toString());
}

Para obter um conjunto completo de exemplos de Java para Confluent em funcionamento, consulte o repositório de extensão do Kafka.

Atributos

As bibliotecas de C# em processo e de processo de trabalho isolado usam o KafkaTriggerAttribute para definir o gatilho da função.

A tabela a seguir explica as propriedades que você pode definir usando este atributo de gatilho:

Parâmetro Descrição
BrokerList (Obrigatório) A lista de agentes do Kafka monitorados pelo gatilho. Consulte Conexões para obter mais informações.
Tópico (Obrigatório) O tópico monitorado pelo gatilho.
ConsumerGroup (Opcional) Grupo de consumidores do Kafka usado pelo gatilho.
AvroSchema (Opcional) Esquema de um registro genérico ao usar o protocolo Avro.
AuthenticationMode (Opcional) O modo de autenticação ao usar a autenticação SASL (Simple Authentication and Security Layer). Os valores compatíveis são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
Nome de usuário (Opcional) O nome de usuário para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
Senha (Opcional) A senha para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
Protocolo (Opcional) O protocolo de segurança usado ao se comunicar com agentes. Os valores compatíveis são plaintext (padrão), ssl, sasl_plaintext, sasl_ssl.
SslCaLocation (Opcional) Caminho para o arquivo de certificado da AC para verificar o certificado do agente.
SslCertificateLocation (Opcional) Caminho para o certificado do cliente.
SslKeyLocation (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
SslKeyPassword (Opcional) Senha do certificado do cliente.

Anotações

A anotação KafkaTrigger permite que você crie uma função executada quando um tópico é recebido. As opções compatíveis incluem os seguintes elementos:

Elemento Descrição
name (Obrigatório) O nome da variável que representa a fila ou mensagem de tópico no código da função.
brokerList (Obrigatório) A lista de agentes do Kafka monitorados pelo gatilho. Consulte Conexões para obter mais informações.
topic (Obrigatório) O tópico monitorado pelo gatilho.
cardinalidade (Opcional) Indica a cardinalidade da entrada do gatilho. Os valores compatíveis são ONE (padrão) e MANY. Use ONE quando a entrada for uma única mensagem e MANY quando a entrada for uma matriz de mensagens. Ao usar MANY, defina também um dataType.
dataType Define como o Functions lida com o valor do parâmetro. Por padrão, o valor é obtido como uma cadeia de caracteres, e o Functions tenta desserializá-la para o POJO (objeto Java). Quando string, a entrada é tratada apenas como uma cadeia de caracteres. Quando binary, a mensagem é recebida como dados binários, e o Functions tenta desserializá-los para um byte de tipo de parâmetro real[].
consumerGroup (Opcional) Grupo de consumidores do Kafka usado pelo gatilho.
avroSchema (Opcional) Esquema de um registro genérico ao usar o protocolo Avro.
authenticationMode (Opcional) O modo de autenticação ao usar a autenticação SASL (Simple Authentication and Security Layer). Os valores compatíveis são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
username (Opcional) O nome de usuário para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
password (Opcional) A senha para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
protocol (Opcional) O protocolo de segurança usado ao se comunicar com agentes. Os valores compatíveis são plaintext (padrão), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Opcional) Caminho para o arquivo de certificado da AC para verificar o certificado do agente.
sslCertificateLocation (Opcional) Caminho para o certificado do cliente.
sslKeyLocation (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
sslKeyPassword (Opcional) Senha do certificado do cliente.

Configuração

A tabela a seguir explica as propriedades de configuração de associação que você define no arquivo function.json.

Propriedade function.json Descrição
tipo (Obrigatório) Deve ser definido como kafkaTrigger.
direction (Obrigatório) Deve ser definido como in.
name (Obrigatório) O nome da variável que representa os dados agenciados no código de função.
brokerList (Obrigatório) A lista de agentes do Kafka monitorados pelo gatilho. Consulte Conexões para obter mais informações.
topic (Obrigatório) O tópico monitorado pelo gatilho.
cardinalidade (Opcional) Indica a cardinalidade da entrada do gatilho. Os valores compatíveis são ONE (padrão) e MANY. Use ONE quando a entrada for uma única mensagem e MANY quando a entrada for uma matriz de mensagens. Ao usar MANY, defina também um dataType.
dataType Define como o Functions lida com o valor do parâmetro. Por padrão, o valor é obtido como uma cadeia de caracteres, e o Functions tenta desserializá-la para o POJO (objeto Java). Quando string, a entrada é tratada apenas como uma cadeia de caracteres. Quando binary, a mensagem é recebida como dados binários, e o Functions tenta desserializá-los para um byte de tipo de parâmetro real[].
consumerGroup (Opcional) Grupo de consumidores do Kafka usado pelo gatilho.
avroSchema (Opcional) Esquema de um registro genérico ao usar o protocolo Avro.
authenticationMode (Opcional) O modo de autenticação ao usar a autenticação SASL (Simple Authentication and Security Layer). Os valores compatíveis são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
username (Opcional) O nome de usuário para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
password (Opcional) A senha para autenticação SASL. Não há suporte quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
protocol (Opcional) O protocolo de segurança usado ao se comunicar com agentes. Os valores compatíveis são plaintext (padrão), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Opcional) Caminho para o arquivo de certificado da AC para verificar o certificado do agente.
sslCertificateLocation (Opcional) Caminho para o certificado do cliente.
sslKeyLocation (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
sslKeyPassword (Opcional) Senha do certificado do cliente.

Uso

Atualmente, há suporte para eventos do Kafka como cadeias de caracteres e matrizes de cadeia de caracteres que são payloads JSON.

As mensagens do Kafka são passadas para a função como cadeias de caracteres e matrizes de cadeia de caracteres que são payloads JSON.

Em um plano Premium, você deve habilitar o monitoramento de escala de runtime para que a saída do Kafka possa ser dimensionada para várias instâncias. Para saber mais, confira Habilitar o dimensionamento de runtime.

Você não pode usar o recurso Testar/Executar da página Código + Teste no Portal do Microsoft Azure para trabalhar com gatilhos do Kafka. Em vez disso, você deve enviar eventos de teste diretamente para o tópico que está sendo monitorado pelo gatilho.

Para obter um conjunto completo de configurações de host.json compatíveis para o gatilho do Kafka, consulte as configurações de host.json.

conexões

Todas as informações obrigatórias exigidas por seus gatilhos e associações devem ser mantidas nas configurações do aplicativo, não nas definições obrigatórias em seu código. Isso é verdadeiro para credenciais, que nunca devem ser armazenadas em seu código.

Importante

As configurações de credencial devem fazer referência a uma configuração de aplicativo. Não embutir credenciais em seu código ou nos arquivos de configuração. Ao executar localmente, use o arquivo local.settings.json para suas credenciais e não publique o arquivo local.settings.json.

Ao se conectar a um cluster do Kafka gerenciado fornecido pelo Confluent no Azure, verifique se as seguintes credenciais de autenticação para seu ambiente do Confluent Cloud estão definidas no gatilho ou na associação:

Setting Valor recomendado Descrição
BrokerList BootstrapServer A configuração de aplicativo nomeada BootstrapServer contém o valor do servidor bootstrap encontrado na página de configurações do Confluent Cloud. O valor é semelhante a xyz-xyzxzy.westeurope.azure.confluent.cloud:9092.
Nome de usuário ConfluentCloudUsername A configuração de aplicativo nomeada ConfluentCloudUsername contém a chave de acesso da API no site do Confluent Cloud.
Senha ConfluentCloudPassword A configuração de aplicativo nomeada ConfluentCloudPassword contém o segredo da API obtido no site do Confluent Cloud.

Os valores de cadeia de caracteres usados para essas configurações devem estar presentes como configurações de aplicativo no Azure ou na coleção Values no arquivo local.settings.json durante o desenvolvimento local.

Você também deve determinar o Protocol, AuthenticationMode e SslCaLocation em suas definições de associação.

Próximas etapas