Триггер Apache Kafka для Функций Azure

  • Статья

Триггер Apache Kafka можно использовать в Функциях Azure для запуска кода функции в ответ на сообщения в разделах Kafka. Вы также можете использовать выходную привязку Kafka для записи из функции в раздел. Сведения о настройке и конфигурации см. в статье Обзор привязок Apache Kafka для Функций Azure.

Внимание

Привязки Kafka доступны только для Функций в составе эластичного плана "Премиум" и плана "Выделенный (Служба приложений)". Они поддерживаются только в среде выполнения Функций версии 3.x и выше.

Пример

Использование триггера зависит от модальности C#, используемой в приложении-функции. Это может быть один из следующих вариантов:

Изолированная библиотека классов рабочих процессов, скомпилированная функция C# выполняется в процессе, изолированном от среды выполнения.

Используемые атрибуты зависят от конкретного поставщика событий.

В следующем примере показана функция C#, которая считывает и регистрирует сообщение Kafka как событие 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"]}");
}

Чтобы получить события в пакете, используйте в качестве входных данных массив строк, как показано в следующем примере:

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

Следующая функция регистрирует сообщение и заголовки для события 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"])}");

    }
}

Полный набор рабочих примеров .NET см. в репозитории расширений Kafka.

Примечание.

Эквивалентный набор примеров TypeScript см. в репозитории расширений Kafka

Конкретные свойства файла function.json зависят от поставщика событий. В данных примерах это Confluent или Центры событий Azure. В следующих примерах показан триггер Kafka для функции, которая считывает и регистрирует сообщение Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

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

Затем при активации функции выполняется следующий код:

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

Чтобы получить события в пакете, задайте для параметра cardinality в файле function.json значение many, как показано в приведенных ниже примерах.

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

Затем следующий код анализирует массив событий и регистрирует данные события:

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

Следующий код также регистрирует данные заголовка:

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

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующий файл function.json определяет триггер для конкретного поставщика, используя универсальную схему 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%"
  } ]
}

Затем при активации функции выполняется следующий код:

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

Полный набор рабочих примеров JavaScript см. в репозитории расширений Kafka.

Конкретные свойства файла function.json зависят от поставщика событий. В данных примерах это Confluent или Центры событий Azure. В следующих примерах показан триггер Kafka для функции, которая считывает и регистрирует сообщение Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

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

Затем при активации функции выполняется следующий код:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

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

Чтобы получить события в пакете, задайте для параметра cardinality в файле function.json значение many, как показано в приведенных ниже примерах.

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

Затем следующий код анализирует массив событий и регистрирует данные события:

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

Следующий код также регистрирует данные заголовка:

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

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующий файл function.json определяет триггер для конкретного поставщика, используя универсальную схему 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%"
  } ]
}

Затем при активации функции выполняется следующий код:

using namespace System.Net

param($kafkaEvent, $TriggerMetadata)

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

Полный набор рабочих примеров PowerShell см. в репозитории расширений Kafka.

Конкретные свойства файла function.json зависят от поставщика событий. В данных примерах это Confluent или Центры событий Azure. В следующих примерах показан триггер Kafka для функции, которая считывает и регистрирует сообщение Kafka.

Следующий файл function.json определяет триггер для конкретного поставщика.

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

Затем при активации функции выполняется следующий код:

import logging
from azure.functions import KafkaEvent

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

Чтобы получить события в пакете, задайте для параметра cardinality в файле function.json значение many, как показано в приведенных ниже примерах.

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

Затем следующий код анализирует массив событий и регистрирует данные события:

import logging
import typing
from azure.functions import KafkaEvent

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

Следующий код также регистрирует данные заголовка:

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

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующий файл function.json определяет триггер для конкретного поставщика, используя универсальную схему 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%"
  } ]
}

Затем при активации функции выполняется следующий код:

import logging
from azure.functions import KafkaEvent

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

Полный набор рабочих примеров Python см. в репозитории расширений Kafka.

Заметки, используемые для настройки триггера, зависят от конкретного поставщика событий.

В следующем примере показана функция Java, которая считывает и регистрирует содержимое события 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);
}

Чтобы получить события в пакете, используйте входную строку в качестве массива, как показано в следующем примере:

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

Следующая функция регистрирует сообщение и заголовки для события 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);                    
                }                
            }
        }

Вы можете определить универсальную схему Avro для события, переданного триггеру. Следующая функция определяет триггер для конкретного поставщика, используя универсальную схему 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());
}

Полный набор рабочих примеров Java для Confluent см. в репозитории расширений Kafka.

Атрибуты

Библиотеки C# в процессе и изолированном рабочем процессе используются KafkaTriggerAttribute для определения триггера функции.

В следующей таблице описаны свойства, которые можно задать с помощью этого атрибута триггера:

Параметр Описание
BrokerList (Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
Раздел (Обязательно) Раздел, отслеживаемый триггером.
ConsumerGroup (Необязательно) Группа потребителей Kafka, используемая триггером.
AvroSchema (Необязательно) Схема универсальной записи при использовании протокола Avro.
AuthenticationMode (Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: Gssapi, Plain (по умолчанию), ScramSha256, ScramSha512.
Username (Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
Пароль (Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
Протокол (Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: plaintext (по умолчанию), ssl, sasl_plaintext, sasl_ssl.
SslCaLocation (Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
SslCertificateLocation (Необязательно) Путь к сертификату клиента.
SslKeyLocation (Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
SslKeyPassword (Необязательно) Пароль для сертификата клиента.

Заметки

Заметка KafkaTrigger позволяет создать функцию, которая выполняется при получении раздела. Поддерживаемые варианты включают следующие элементы:

Элемент Description
name (Обязательно) Имя переменной, представляющей сообщение очереди или раздела в коде функции.
brokerList (Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
topic (Обязательно) Раздел, отслеживаемый триггером.
кратность (Необязательно) Указывает кратность входных данных триггера. Поддерживаемые значения: ONE (по умолчанию) и MANY. Используйте ONE, если входные данные являются одним сообщением, и MANY, если входные данные являются массивом сообщений. При использовании MANY необходимо также задать параметр dataType.
dataType Определяет, как Функции обрабатывают значение параметра. По умолчанию полученное значение представляет собой строку, и Функции пытаются десериализовать ее до объекта POJO. При string входные данные обрабатываются просто как строка. При binary сообщение приходит в формате двоичных данных и Функции пытаются десериализовать его до фактического типа параметра byte[].
consumerGroup (Необязательно) Группа потребителей Kafka, используемая триггером.
avroSchema (Необязательно) Схема универсальной записи при использовании протокола Avro.
authenticationMode (Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: Gssapi, Plain (по умолчанию), ScramSha256, ScramSha512.
username (Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
пароль (Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
protocol (Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: plaintext (по умолчанию), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
sslCertificateLocation (Необязательно) Путь к сертификату клиента.
sslKeyLocation (Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
sslKeyPassword (Необязательно) Пароль для сертификата клиента.

Настройка

В следующей таблице описываются свойства конфигурации привязки, которые задаются в файле function.json.

Свойство в function.json Описание
type Обязательное. Необходимо задать значение kafkaTrigger.
direction Обязательное. Необходимо задать значение in.
name (Обязательно) Имя переменной, представляющей данные, полученные через брокер, в коде функции.
brokerList (Обязательно) Список брокеров Kafka, отслеживаемых триггером. Дополнительные сведения см. в разделе Подключения.
topic (Обязательно) Раздел, отслеживаемый триггером.
кратность (Необязательно) Указывает кратность входных данных триггера. Поддерживаемые значения: ONE (по умолчанию) и MANY. Используйте ONE, если входные данные являются одним сообщением, и MANY, если входные данные являются массивом сообщений. При использовании MANY необходимо также задать параметр dataType.
dataType Определяет, как Функции обрабатывают значение параметра. По умолчанию полученное значение представляет собой строку, и Функции пытаются десериализовать ее до объекта POJO. При string входные данные обрабатываются просто как строка. При binary сообщение приходит в формате двоичных данных и Функции пытаются десериализовать его до фактического типа параметра byte[].
consumerGroup (Необязательно) Группа потребителей Kafka, используемая триггером.
avroSchema (Необязательно) Схема универсальной записи при использовании протокола Avro.
authenticationMode (Необязательно) Режим, используемый при простой проверке подлинности и проверке подлинности уровня безопасности (SASL). Поддерживаемые значения: Gssapi, Plain (по умолчанию), ScramSha256, ScramSha512.
username (Необязательно) Имя пользователя для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
пароль (Необязательно) Пароль для проверки подлинности SASL. Не поддерживается, если AuthenticationMode имеет значение Gssapi. Дополнительные сведения см. в разделе Подключения.
protocol (Необязательно) Протокол безопасности, используемый при взаимодействии с брокерами. Поддерживаемые значения: plaintext (по умолчанию), ssl, sasl_plaintext, sasl_ssl.
sslCaLocation (Необязательно) Путь к файлу сертификата ЦС для проверки сертификата брокера.
sslCertificateLocation (Необязательно) Путь к сертификату клиента.
sslKeyLocation (Необязательно) Путь к закрытому ключу клиента (PEM), используемому для проверки подлинности.
sslKeyPassword (Необязательно) Пароль для сертификата клиента.

Использование

События Kafka в настоящее время поддерживаются как строки и массивы строк, которые являются полезными данными JSON.

Сообщения Kafka передаются функции в виде строк и массивов строк, которые являются полезными данными JSON.

В плане "Премиум" необходимо включить мониторинг масштабирования среды выполнения для выходных данных Kafka, чтобы иметь возможность горизонтально увеличить масштаб до нескольких экземпляров. Дополнительные сведения см. в статье Включение масштабирования среды выполнения.

Для работы с триггерами Kafka нельзя использовать функцию "Тест и запуск" страницы "Код и тест" на портале Azure. Вместо этого необходимо отправлять тестовые события непосредственно в раздел, отслеживаемый триггером.

Полный набор поддерживаемых параметров host.json для триггера Kafka см. в статье Параметры host.json.

Связи

Все сведения о подключении, необходимые триггерам и привязкам, должны храниться в параметрах приложения, а не в определениях привязок в коде. Это верно для учетных данных, которые никогда не должны храниться в коде.

Внимание

Параметры учетных данных должны ссылаться на параметр приложения. Не прописывайте учетные данные в файлах кода или конфигурации. При локальном запуске используйте файл local.settings.json для учетных данных и не публикуйте его.

При подключении к управляемому кластеру Kafka, предоставляемому Confluent в Azure, убедитесь, что в триггере или привязке заданы следующие учетные данные проверки подлинности для среды Confluent Cloud:

Параметр Рекомендуемое значение Description
BrokerList BootstrapServer Параметр приложения с именем BootstrapServer содержит значение сервера начальной загрузки, найденное на странице параметров Confluent Cloud. Значение напоминает xyz-xyzxzy.westeurope.azure.confluent.cloud:9092.
Username ConfluentCloudUsername Параметр приложения с именем ConfluentCloudUsername содержит ключ доступа API, полученный от веб-сайта Confluent Cloud.
Пароль ConfluentCloudPassword Параметр приложения с именем ConfluentCloudPassword содержит секрет API, полученный от веб-сайта Confluent Cloud.

Строковые значения, используемые для этих параметров, должны присутствовать в качестве параметров приложения в Azure или коллекции Values в файле local.settings.json во время локальной разработки.

Необходимо также задать Protocol, AuthenticationMode и SslCaLocation в определениях привязок.

Следующие шаги