Как подключиться к EventHubs с расширением Apache -Kafka, используя функции Azure?

Question

Могу ли я использовать SharedAccessKey для подключения к брокеру (EventHubs)?

Мне не удается подключиться к моим Azure EventHubs.

Мы используем SharedAccessKey вместо SSL для подключения и у меня есть эта конфигурация, чтобы сделать это.

"EventBusConfig": {

    "BootstrapServers": "anyname.servicebus.windows.net:9093",

    "SecurityProtocol": "SaslSsl",

    "SaslMechanism": "Plain",

    "SaslUsername": "$ConnectionString",

    "SaslPassword": 
    "Endpoint=sb://anyname.servicebus.windows.net/;SharedAccessKeyName=anyname.;SharedAccessKey=CtDbJ/Kfjs749
    8s--anypassword--SkSk749/z2Z5Fr9///33/qQ+R6Cyg=",

    "SocketTimeoutMs": "60000",

    "SessionTimeoutMs": "30000",

    "GroupId": "NameOfTheGroup",

    "AutoOffsetReset": "Earliest",

    "BrokerVersionFallback": "1.0.0",

    "Debug": "cgrp"
}

Но, похоже, мне нужен путь сертификации (файл pem)

Я хочу создать простое сообщение, подобное этому

Я использую https://github.com/Azure/azure-functions-kafka-extension, но я не знаю, может ли эта бета-библиотека обрабатывать SharedAccessKey.

Я получил эту ошибку при попытке подключения:

Любая помощь будет оценена

Answer 1

Мне удалось создавать и использовать сообщения, используя расширение "https://github.com/Azure/azure-functions-kafka-extension".

Использовать сообщение было легко, поскольку свойство "EventHubConnectionString" очень интуитивно понятно.

Чтобы создать сообщение, вам необходимо настроить сертификат CA, я подумал, что мне нужно это из Azure, но я ошибся и просто следовал этим инструкциям, чтобы заставить его работать.

Загрузите и установите местоположение сертификации CA. Как описано в документации Confluent, библиотека. NET не имеет возможности доступа к root сертификатам CA. Пропуск этого шага приведет к тому, что ваша функция выдаст ошибку "sasl_ssl: //xyz-xyzxzy.westeurope.azure.confluent.cloud: 9092 / bootstrap: не удалось проверить сертификат брокера: невозможно получить сертификат локального эмитента (после 135 мс в состоянии ПОДКЛЮЧИТЬСЯ) "

Чтобы преодолеть это, нам необходимо:

• Скачать сертификат CA (т.е. из https://curl.haxx.se/ca/cacert.pem).
• Переименуйте файл сертификата во что угодно, кроме cacert.pem, чтобы избежать конфликта с существующим сертификатом EventHubs Kafka, который является частью расширения.
• Включите файл в проект, установив «копировать в выходной каталог»
• Установить свойство атрибута триггера SslCaLocation. В этом примере мы установили значение confluent_cloud_cacert.pem

Это моя функция Azure моего производителя с привязкой Kafka

using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using Microsoft.Azure.WebJobs.Extensions.Kafka;

namespace EY.Disruptor.AzureFunctionsWithKafka
{
public static class Function
{
    [FunctionName("Producer")]
    public static async Task<IActionResult> Run(
            [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)] HttpRequest req,
            [Kafka("BootstrapServer",
            "topic.event",
            Username = "ConfluentCloudUsername",
            Password = "ConfluentCloudPassword",
            SslCaLocation = "confluent_cloud_cacert.pem",
            AuthenticationMode = BrokerAuthenticationMode.Plain,
            Protocol = BrokerProtocol.SaslSsl
        )] IAsyncCollector<KafkaEventData<string>> events,
            ILogger log)
    {
        log.LogInformation("C# HTTP trigger function processed a request.");

        string name = req.Query["name"];

        string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
        dynamic data = JsonConvert.DeserializeObject(requestBody);
        name ??= data?.name;

        string responseMessage = string.IsNullOrEmpty(name)
            ? "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response."
            : $"Hello, {name}. This HTTP triggered function executed successfully.";

        var kafkaEvent = new KafkaEventData<string>()
        {
            Value = name
        };

        await events.AddAsync(kafkaEvent);

        return new OkObjectResult(responseMessage);
    }
}
}

Это моя функция потребления Azure с привязкой Кафки

using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Kafka;
using Microsoft.Extensions.Logging;

namespace EY.Disruptor.AzureFunctionsWithKafka
{
public static class Consumer
{
    [FunctionName("FunctionKafkaConsumer")]
    public static void Run(
        [KafkaTrigger("BootstrapServer",
        "topic.event",
        Username = "ConfluentCloudUsername",
        Password = "ConfluentCloudPassword",
        EventHubConnectionString = "ConfluentCloudPassword",
        AuthenticationMode = BrokerAuthenticationMode.Plain,
        Protocol = BrokerProtocol.SaslSsl,
        ConsumerGroup = "Group1")] KafkaEventData<string>[] kafkaEvents,
        ILogger logger)
    {
        foreach (var kafkaEvent in kafkaEvents)
        {
            logger.LogInformation(kafkaEvent.Value);
        }
    }
}
}

Это мои local.settings. json

{
  "IsEncrypted": false,
  "Values": {
     "AzureWebJobsStorage": "UseDevelopmentStorage=true",
     "FUNCTIONS_WORKER_RUNTIME": "dotnet",
     "BootstrapServer": "zyxabc.servicebus.windows.net:9093",
     "ConfluentCloudUsername": "$ConnectionString",
     "ConfluentCloudPassword": "Endpoint=sb://zyxabc.servicebus.windows.net/;SharedAccessKeyName=TestSvc;SharedAccessKey=YAr/="
   }
}

И конечно же инициализация в Startup.cs

public void Configure(IWebJobsBuilder builder)
{
   builder.AddKafka();
}

Я надеюсь, что эта рекомендация помогает другим людям:)