Trigger Apache Kafka pro Azure Functions

  • Článek

Trigger Apache Kafka ve službě Azure Functions můžete použít ke spuštění kódu funkce v reakci na zprávy v tématech Kafka. Výstupní vazbu Kafka můžete použít také k zápisu z funkce do tématu. Informace o nastavení a konfiguraci najdete v přehledu vazeb Apache Kafka pro Azure Functions.

Důležité

Vazby Kafka jsou dostupné jenom pro funkce v plánu Elastic Premium a vyhrazeném plánu (App Service). Podporují se pouze ve verzi 3.x a novější verzi modulu runtime Functions.

Příklad

Použití triggeru závisí na způsobu použití jazyka C# používaném ve vaší aplikaci funkcí, což může být jeden z následujících režimů:

Kompilovaná funkce C# v izolované knihovně tříd pracovních procesů běží v procesu izolovaném od modulu runtime.

Atributy, které použijete, závisí na konkrétním poskytovateli událostí.

Následující příklad ukazuje funkci jazyka C#, která čte a protokoluje zprávu Kafka jako událost 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"]}");
}

Pokud chcete přijímat události v dávce, použijte jako vstup pole řetězců, jak je znázorněno v následujícím příkladu:

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

Následující funkce protokoluje zprávu a hlavičky pro událost 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"])}");

    }
}

Kompletní sadu funkčních příkladů .NET najdete v úložišti rozšíření Kafka.

Poznámka:

Ekvivalentní sadu příkladů TypeScriptu najdete v úložišti rozšíření Kafka.

Konkrétní vlastnosti souboru function.json závisí na poskytovateli událostí, který v těchto příkladech jsou Confluent nebo Azure Event Hubs. Následující příklady ukazují trigger Kafka pro funkci, která čte a protokoluje zprávu Kafka.

Následující function.json definuje aktivační událost pro konkrétního poskytovatele:

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

Následující kód se pak spustí při aktivaci funkce:

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

Pokud chcete přijímat události v dávce, nastavte cardinality hodnotu v many souboru function.json, jak je znázorněno v následujících příkladech:

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

Následující kód pak parsuje pole událostí a protokoluje data události:

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);
};

Následující kód také protokoluje data hlaviček:

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);
};

Pro událost předanou triggeru můžete definovat obecné schéma Avro. Následující function.json definuje aktivační událost pro konkrétního zprostředkovatele s obecným schématem Avro:

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

Následující kód se pak spustí při aktivaci funkce:

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

Kompletní sadu funkčních příkladů JavaScriptu najdete v úložišti rozšíření Kafka.

Konkrétní vlastnosti souboru function.json závisí na poskytovateli událostí, který v těchto příkladech jsou Confluent nebo Azure Event Hubs. Následující příklady ukazují trigger Kafka pro funkci, která čte a protokoluje zprávu Kafka.

Následující function.json definuje aktivační událost pro konkrétního poskytovatele:

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

Následující kód se pak spustí při aktivaci funkce:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

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

Pokud chcete přijímat události v dávce, nastavte cardinality hodnotu v many souboru function.json, jak je znázorněno v následujících příkladech:

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

Následující kód pak parsuje pole událostí a protokoluje data události:

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"
}

Následující kód také protokoluje data hlaviček:

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"
    }
}

Pro událost předanou triggeru můžete definovat obecné schéma Avro. Následující function.json definuje aktivační událost pro konkrétního zprostředkovatele s obecným schématem Avro:

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

Následující kód se pak spustí při aktivaci funkce:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

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

Kompletní sadu funkčních příkladů PowerShellu najdete v úložišti rozšíření Kafka.

Konkrétní vlastnosti souboru function.json závisí na poskytovateli událostí, který v těchto příkladech jsou Confluent nebo Azure Event Hubs. Následující příklady ukazují trigger Kafka pro funkci, která čte a protokoluje zprávu Kafka.

Následující function.json definuje aktivační událost pro konkrétního poskytovatele:

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

Následující kód se pak spustí při aktivaci funkce:

import logging
from azure.functions import KafkaEvent

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

Pokud chcete přijímat události v dávce, nastavte cardinality hodnotu v many souboru function.json, jak je znázorněno v následujících příkladech:

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

Následující kód pak parsuje pole událostí a protokoluje data události:

import logging
import typing
from azure.functions import KafkaEvent

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

Následující kód také protokoluje data hlaviček:

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')))

Pro událost předanou triggeru můžete definovat obecné schéma Avro. Následující function.json definuje aktivační událost pro konkrétního zprostředkovatele s obecným schématem Avro:

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

Následující kód se pak spustí při aktivaci funkce:

import logging
from azure.functions import KafkaEvent

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

Kompletní sadu funkčních příkladů Pythonu najdete v úložišti rozšíření Kafka.

Poznámky, které použijete ke konfiguraci triggeru, závisí na konkrétním poskytovateli událostí.

Následující příklad ukazuje funkci Java, která čte a protokoluje obsah události 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);
}

Pokud chcete přijímat události v dávce, použijte vstupní řetězec jako pole, jak je znázorněno v následujícím příkladu:

@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);
        }    
}

Následující funkce protokoluje zprávu a hlavičky pro událost 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);                    
                }                
            }
        }

Pro událost předanou triggeru můžete definovat obecné schéma Avro. Následující funkce definuje aktivační událost pro konkrétního zprostředkovatele s obecným schématem Avro:

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());
}

Kompletní sadu funkčních příkladů Javy pro Confluent najdete v úložišti rozšíření Kafka.

Atributy

Knihovny C# v procesu i izolovaného pracovního procesu používají KafkaTriggerAttribute k definování triggeru funkce.

Následující tabulka vysvětluje vlastnosti, které můžete nastavit pomocí tohoto atributu triggeru:

Parametr Popis
BrokerList (Povinné) Seznam zprostředkovatelů Kafka monitorovaných triggerem Další informace najdete v tématu Připojení .
Předmět zájmu (Povinné) Téma monitorované triggerem.
Skupina příjemců (Volitelné) Skupina příjemců Kafka používaná triggerem
AvroSchema (Volitelné) Schéma obecného záznamu při použití protokolu Avro
AuthenticationMode (Volitelné) Režim ověřování při použití ověřování SASL (Simple Authentication and Security Layer). Podporované hodnoty jsou Gssapi, Plain (výchozí), ScramSha256, ScramSha512.
Uživatelské jméno (Volitelné) Uživatelské jméno pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
Heslo (Volitelné) Heslo pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
Protokol (Volitelné) Protokol zabezpečení používaný při komunikaci s zprostředkovateli. Podporované hodnoty jsou plaintext (výchozí), ssl, sasl_plaintext, sasl_ssl.
SslCaLocation (Volitelné) Cesta k souboru certifikátu certifikační autority pro ověření certifikátu zprostředkovatele
SslCertificateLocation (Volitelné) Cesta k certifikátu klienta
SslKeyLocation (Volitelné) Cesta k privátnímu klíči klienta (PEM) používanému k ověřování
SslKeyPassword (Volitelné) Heslo pro certifikát klienta.

Poznámky

Poznámka KafkaTrigger umožňuje vytvořit funkci, která se spustí při přijetí tématu. Mezi podporované možnosti patří následující prvky:

Element (Prvek) Popis
Jméno (Povinné) Název proměnné, která představuje zprávu fronty nebo tématu v kódu funkce.
brokerList (Povinné) Seznam zprostředkovatelů Kafka monitorovaných triggerem Další informace najdete v tématu Připojení .
topic (Povinné) Téma monitorované triggerem.
mohutnost (Volitelné) Určuje kardinalitu vstupu triggeru. Podporované hodnoty jsou ONE (výchozí) a MANY. Použijte ONE , když je vstup jedinou zprávou a MANY když je vstup polem zpráv. Při použití MANYmusíte také nastavit dataType.
Datatype Definuje, jak služba Functions zpracovává hodnotu parametru. Ve výchozím nastavení je hodnota získána jako řetězec a Functions se pokusí deserializovat řetězec na skutečný prostý starý java objekt (POJO). Když stringse vstup považuje za řetězec. Když binaryse zpráva přijme jako binární data a functions se pokusí deserializovat na skutečný typ parametru bajt[].
consumerGroup (Volitelné) Skupina příjemců Kafka používaná triggerem
avroSchema (Volitelné) Schéma obecného záznamu při použití protokolu Avro
authenticationMode (Volitelné) Režim ověřování při použití ověřování SASL (Simple Authentication and Security Layer). Podporované hodnoty jsou Gssapi, Plain (výchozí), ScramSha256, ScramSha512.
uživatelské jméno (Volitelné) Uživatelské jméno pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
heslo (Volitelné) Heslo pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
protokol (Volitelné) Protokol zabezpečení používaný při komunikaci s zprostředkovateli. Podporované hodnoty jsou plaintext (výchozí), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Volitelné) Cesta k souboru certifikátu certifikační autority pro ověření certifikátu zprostředkovatele
sslCertificateLocation (Volitelné) Cesta k certifikátu klienta
sslKeyLocation (Volitelné) Cesta k privátnímu klíči klienta (PEM) používanému k ověřování
sslKeyPassword (Volitelné) Heslo pro certifikát klienta.

Konfigurace

Následující tabulka vysvětluje vlastnosti konfigurace vazby, které jste nastavili v souboru function.json .

vlastnost function.json Popis
type (Povinné) Musí být nastavena na kafkaTriggerhodnotu .
direction (Povinné) Musí být nastavena na inhodnotu .
Jméno (Povinné) Název proměnné, která představuje zprostředkovaná data v kódu funkce.
brokerList (Povinné) Seznam zprostředkovatelů Kafka monitorovaných triggerem Další informace najdete v tématu Připojení .
topic (Povinné) Téma monitorované triggerem.
mohutnost (Volitelné) Určuje kardinalitu vstupu triggeru. Podporované hodnoty jsou ONE (výchozí) a MANY. Použijte ONE , když je vstup jedinou zprávou a MANY když je vstup polem zpráv. Při použití MANYmusíte také nastavit dataType.
Datatype Definuje, jak služba Functions zpracovává hodnotu parametru. Ve výchozím nastavení je hodnota získána jako řetězec a Functions se pokusí deserializovat řetězec na skutečný prostý starý java objekt (POJO). Když stringse vstup považuje za řetězec. Když binaryse zpráva přijme jako binární data a functions se pokusí deserializovat na skutečný typ parametru bajt[].
consumerGroup (Volitelné) Skupina příjemců Kafka používaná triggerem
avroSchema (Volitelné) Schéma obecného záznamu při použití protokolu Avro
authenticationMode (Volitelné) Režim ověřování při použití ověřování SASL (Simple Authentication and Security Layer). Podporované hodnoty jsou Gssapi, Plain (výchozí), ScramSha256, ScramSha512.
uživatelské jméno (Volitelné) Uživatelské jméno pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
heslo (Volitelné) Heslo pro ověřování SASL. Nepodporuje se, pokud AuthenticationMode je .Gssapi Další informace najdete v tématu Připojení .
protokol (Volitelné) Protokol zabezpečení používaný při komunikaci s zprostředkovateli. Podporované hodnoty jsou plaintext (výchozí), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Volitelné) Cesta k souboru certifikátu certifikační autority pro ověření certifikátu zprostředkovatele
sslCertificateLocation (Volitelné) Cesta k certifikátu klienta
sslKeyLocation (Volitelné) Cesta k privátnímu klíči klienta (PEM) používanému k ověřování
sslKeyPassword (Volitelné) Heslo pro certifikát klienta.

Využití

Události Kafka se v současné době podporují jako řetězce a pole řetězců, které jsou datovými částmi JSON.

Zprávy Kafka se funkci předávají jako řetězce a pole řetězců, které jsou datovými částmi JSON.

V plánu Premium musíte povolit monitorování škálování modulu runtime pro výstup Kafka, aby bylo možné škálovat na více instancí. Další informace najdete v tématu Povolení škálování modulu runtime.

K práci s triggery Kafka nemůžete použít funkci Test/Spustit na stránce Kód + Test na webu Azure Portal. Místo toho je nutné odesílat testovací události přímo do tématu monitorovaného triggerem.

Úplnou sadu podporovaných nastavení host.json pro trigger Kafka najdete v tématu host.json nastavení.

Propojení

Všechny informace o připojení vyžadované triggery a vazbami by se měly udržovat v nastavení aplikace, a ne v definicích vazeb v kódu. To platí pro přihlašovací údaje, které by nikdy neměly být uloženy ve vašem kódu.

Důležité

Nastavení přihlašovacích údajů musí odkazovat na nastavení aplikace. Nezakódujte v kódu ani v konfiguračních souborech pevně zakódujte přihlašovací údaje. Při místním spuštění použijte local.settings.json soubor pro přihlašovací údaje a nepublikujte local.settings.json soubor.

Při připojování ke spravovanému clusteru Kafka poskytovanému Confluentem v Azure se ujistěte, že ve vašem triggeru nebo vazbě jsou nastavené následující ověřovací přihlašovací údaje pro vaše prostředí Confluent Cloud:

Nastavení Doporučená hodnota Popis
BrokerList BootstrapServer Nastavení aplikace s názvem BootstrapServer obsahuje hodnotu serveru bootstrap nalezené na stránce nastavení confluent cloudu. Hodnota se podobá xyz-xyzxzy.westeurope.azure.confluent.cloud:9092.
Uživatelské jméno ConfluentCloudUsername Nastavení aplikace s názvem ConfluentCloudUsername obsahuje přístupový klíč rozhraní API z webu Confluent Cloud.
Heslo ConfluentCloudPassword Nastavení aplikace s názvem ConfluentCloudPassword obsahuje tajný klíč rozhraní API získaný z webu Confluent Cloud.

Řetězcové hodnoty, které použijete pro tato nastavení, musí být přítomné jako nastavení aplikace v Azure nebo v Values kolekci v souboru local.settings.json během místního vývoje.

Měli byste také nastavit , ProtocolAuthenticationModea SslCaLocation v definicích vazeb.

Další kroky