Gatilho Apache Kafka para Azure Functions

  • Artigo

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

Importante

As ligações Kafka só estão disponíveis para Funções no plano Elastic Premium e no plano Dedicado (Serviço de Aplicativo). Eles só são suportados na versão 3.x e versão posterior do tempo de execução do Functions.

Exemplo

O uso do gatilho depende da modalidade C# usada em seu aplicativo de função, que pode ser um dos seguintes modos:

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

Os atributos que você usa dependem do provedor de eventos específico.

O exemplo a seguir mostra uma função C# que lê e registra a mensagem Kafka como um evento 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 no exemplo a seguir:

[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 função a seguir registra a mensagem e os cabeçalhos para o evento 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 .NET de trabalho, consulte o repositório de extensão Kafka.

Nota

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

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

O function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 cardinality valor como many no arquivo function.json, conforme mostrado nos exemplos a seguir:

{
    "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 código a seguir 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 código a seguir também registra os dados do 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);
};

Você pode definir um esquema Avro genérico para o evento passado para o gatilho. A function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 JavaScript de trabalho, consulte o repositório de extensões Kafka.

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

O function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 cardinality valor como many no arquivo function.json, conforme mostrado nos exemplos a seguir:

{
    "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 código a seguir 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 código a seguir também registra os dados do 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"
    }
}

Você pode definir um esquema Avro genérico para o evento passado para o gatilho. A function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 Kafka.

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

O function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 cardinality valor como many no arquivo function.json, conforme mostrado nos exemplos a seguir:

{
      "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 código a seguir 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 código a seguir também registra os dados do 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')))

Você pode definir um esquema Avro genérico para o evento passado para o gatilho. A function.json a seguir 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 código a seguir é executado quando a função é acionada:

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 Python de trabalho, consulte o repositório de extensão Kafka.

As anotações usadas 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 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 no exemplo a seguir:

@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 função a seguir registra a mensagem e os cabeçalhos para o evento 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);                    
                }                
            }
        }

Você pode definir um esquema Avro genérico para o evento passado para o gatilho. A função a seguir define um 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 Java de trabalho para Confluent, consulte o repositório de extensão Kafka.

Atributos

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

A tabela a seguir explica as propriedades que você pode definir usando esse atributo trigger:

Parâmetro Description
Lista de Corretores (Obrigatório) A lista de corretores Kafka monitorados pelo gatilho. Consulte Conexões para obter mais informações.
Tópico (Obrigatório) O tópico monitorado pelo gatilho.
Grupo de Consumidores (Opcional) Grupo de consumidores Kafka usado pelo gatilho.
AvroSchema (Opcional) Esquema de um registro genérico ao usar o protocolo Avro.
Modo de autenticação (Opcional) O modo de autenticação ao usar a autenticação SASL (Simple Authentication and Security Layer). Os valores suportados são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
Nome de utilizador (Opcional) O nome de usuário para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
Palavra-passe (Opcional) A senha para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
Protocolo (Opcional) O protocolo de segurança usado na comunicação com corretores. Os valores suportados são plaintext (padrão), ssl, , sasl_plaintextsasl_ssl.
Localização de SslCa. (Opcional) Caminho para o arquivo de certificado da autoridade de certificação para verificar o certificado do corretor.
SslCertificateLocation (Opcional) Caminho para o certificado do cliente.
SslKeyLocalização (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
SslKeyPassword (Opcional) Senha para o certificado do cliente.

Anotações

A KafkaTrigger anotação permite criar uma função que é executada quando um tópico é recebido. As opções suportadas incluem os seguintes elementos:

Elemento Description
Designação (Obrigatório) O nome da variável que representa a fila ou a mensagem de tópico no código da função.
Lista de corretores (Obrigatório) A lista de corretores 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 suportados são ONE (padrão) e MANY. Use ONE quando a entrada é uma única mensagem e MANY quando a entrada é uma matriz de mensagens. Quando você usa MANYo , você também deve definir um dataTypearquivo .
Tipo de dados Define como Functions lida com o valor do parâmetro. Por padrão, o valor é obtido como uma string e Functions tenta desserializar a string para o objeto Java plain-old real (POJO). Quando string, a entrada é tratada como apenas uma cadeia de caracteres. Quando binary, a mensagem é recebida como dados binários e o Functions tenta desserializá-la para um byte do tipo parâmetro real[].
Grupo de consumidores (Opcional) Grupo de consumidores 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 suportados são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
nome de utilizador (Opcional) O nome de usuário para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
palavra-passe (Opcional) A senha para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
protocolo (Opcional) O protocolo de segurança usado na comunicação com corretores. Os valores suportados são plaintext (padrão), ssl, , sasl_plaintextsasl_ssl.
sslCaLocalização (Opcional) Caminho para o arquivo de certificado da autoridade de certificação para verificar o certificado do corretor.
sslCertificateLocation (Opcional) Caminho para o certificado do cliente.
sslKeyLocalização (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
sslKeyPassword (Opcional) Senha para o certificado do cliente.

Configuração

A tabela a seguir explica as propriedades de configuração de associação definidas no arquivo function.json .

function.json propriedade Description
type (Obrigatório) Deve ser definido como kafkaTrigger.
direção (Obrigatório) Deve ser definido como in.
Designação (Obrigatório) O nome da variável que representa os dados intermediados no código da função.
Lista de corretores (Obrigatório) A lista de corretores 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 suportados são ONE (padrão) e MANY. Use ONE quando a entrada é uma única mensagem e MANY quando a entrada é uma matriz de mensagens. Quando você usa MANYo , você também deve definir um dataTypearquivo .
Tipo de dados Define como Functions lida com o valor do parâmetro. Por padrão, o valor é obtido como uma string e Functions tenta desserializar a string para o objeto Java plain-old real (POJO). Quando string, a entrada é tratada como apenas uma cadeia de caracteres. Quando binary, a mensagem é recebida como dados binários e o Functions tenta desserializá-la para um byte do tipo parâmetro real[].
Grupo de consumidores (Opcional) Grupo de consumidores 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 suportados são Gssapi, Plain (padrão), ScramSha256, ScramSha512.
nome de utilizador (Opcional) O nome de usuário para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
palavra-passe (Opcional) A senha para autenticação SASL. Não suportado quando AuthenticationMode é Gssapi. Consulte Conexões para obter mais informações.
protocolo (Opcional) O protocolo de segurança usado na comunicação com corretores. Os valores suportados são plaintext (padrão), ssl, , sasl_plaintextsasl_ssl.
sslCaLocalização (Opcional) Caminho para o arquivo de certificado da autoridade de certificação para verificar o certificado do corretor.
sslCertificateLocation (Opcional) Caminho para o certificado do cliente.
sslKeyLocalização (Opcional) Caminho para a chave privada do cliente (PEM) usada para autenticação.
sslKeyPassword (Opcional) Senha para o certificado do cliente.

Utilização

Atualmente, há suporte para eventos Kafka como strings e strings arrays que são cargas úteis JSON.

As mensagens Kafka são passadas para a função como strings e matrizes de string que são cargas úteis JSON.

Em um plano Premium, você deve habilitar o monitoramento de escala de tempo de execução para que a saída Kafka possa ser dimensionada para várias instâncias. Para saber mais, consulte Habilitar o dimensionamento em tempo de execução.

Você não pode usar o recurso Test/Run da página Código + Teste no Portal do Azure para trabalhar com gatilhos 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 suportadas para o gatilho Kafka, consulte host.json configurações.

Ligações

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

Importante

As configurações de credenciais devem fazer referência a uma configuração de aplicativo. Não codifice credenciais em seu código ou 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 conectar-se a um cluster Kafka gerenciado fornecido pelo Confluent no Azure, certifique-se de que as seguintes credenciais de autenticação para seu ambiente Confluent Cloud estejam definidas em seu gatilho ou associação:

Definição Valor recomendado Description
Lista de Corretores BootstrapServer A configuração do aplicativo nomeada BootstrapServer contém o valor do servidor de inicialização encontrado na página Configurações do Confluent Cloud. O valor é semelhante a xyz-xyzxzy.westeurope.azure.confluent.cloud:9092.
Nome de utilizador ConfluentCloudUsername A configuração do aplicativo nomeada ConfluentCloudUsername contém a chave de acesso da API do site Confluent Cloud.
Palavra-passe ConfluentCloudPassword A configuração do aplicativo nomeada ConfluentCloudPassword contém o segredo da API obtido do site Confluent Cloud.

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

Você também deve definir o Protocol, AuthenticationModee SslCaLocation em suas definições de vinculação.

Próximos passos