c# - Azure EventHubs 抛出异常 : At least one receiver for the endpoint is created with epoch of '0' ,，因此不允许使用非纪元接收器

Question

简介

大家好，我们目前正在开发一个微服务平台，该平台使用 Azure EventHub 和事件在服务之间发送数据。我们将这些服务命名为:CustomerService、OrderService 和 MobileBFF。

CustomerService 主要发送更新(带有事件)，然后由 OrderService 和 MobileBFF 存储这些更新，以便能够响应查询，而无需调用 CustomerService 获取此数据。

所有这 3 项服务 + 我们在 DEV 环境中的开发人员都使用相同的 ConsumerGroup 连接到这些事件中心。

我们目前仅使用 1 个分区，但计划稍后扩展到多个分区。 (您可以看到我们的代码已经可以从多个分区读取)

异常

不过，我们时不时地会遇到异常(如果它启动，通常会持续抛出此错误一个小时左右)。不过，目前我们仅在开发/测试环境中看到此错误。

异常(exception):

Azure.Messaging.EventHubs.EventHubsException(ConsumerDisconnected): At least one receiver for the endpoint is created with epoch of '0', and so non-epoch receiver is not allowed. Either reconnect with a higher epoch, or make sure all epoch receivers are closed or disconnected.

EventHub 的所有使用者都将其 SequenceNumber 存储在自己的数据库中。这允许我们让每个消费者单独消费事件，并将最后处理的 SequenceNumber 存储在它自己的 SQL 数据库中。当服务(重新)启动时，它从数据库加载 SequenceNumber，然后从这里开始请求事件，直到找不到更多事件。然后它会休眠 100 毫秒，然后重试。这是(稍微简化的)代码:

var consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
string[] allPartitions = null;
await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))
{
    allPartitions = await consumer.GetPartitionIdsAsync(stoppingToken);
}

var allTasks = new List<Task>();

foreach (var partitionId in allPartitions)
{
    //This is required if you reuse variables inside a Task.Run();
    var partitionIdInternal = partitionId;

    allTasks.Add(Task.Run(async () =>
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))
                {
                    EventPosition startingPosition;
                    using (var testScope = _serviceProvider.CreateScope())
                    {
                        var messageProcessor = testScope.ServiceProvider.GetService<EventHubInboxManager<T, EH>>();
                        //Obtains starting position from the database or sets to "Earliest" or "Latest" based on configuration
                        startingPosition = await messageProcessor.GetStartingPosition(_inboxOptions.InboxIdentifier, partitionIdInternal);
                    }

                    while (!stoppingToken.IsCancellationRequested)
                    {
                        bool processedSomething = false;
                        await foreach (PartitionEvent partitionEvent in consumer.ReadEventsFromPartitionAsync(partitionIdInternal, startingPosition, stoppingToken))
                        {
                            processedSomething = true;

                            startingPosition = await messageProcessor.Handle(partitionEvent);
                        }

                        if (processedSomething == false)
                        {
                            await Task.Delay(100, stoppingToken);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                //Log error / delay / retry
            }

        }
    }
}

异常在以下行抛出:

await using (var consumer = new EventHubConsumerClient(consumerGroup, _inboxOptions.EventHubConnectionString, _inboxOptions.EventHubName))

更多调查

上述代码在微服务中运行(在 Azure 中作为 AppService 托管)

接下来，我们还运行 1 个 Azure Function，它也从 EventHub 读取事件。 (可能使用相同的消费者组)。

根据此处的文档:https://learn.microsoft.com/en-us/azure/event-hubs/event-hubs-features#consumer-groups每个消费者组应该可以有 5 个消费者。似乎建议只拥有一个，但我们不清楚如果我们不遵循这一指导会发生什么。

我们确实做了一些测试，手动生成读取事件的服务的多个实例，当超过 5 个实例时，这会导致不同的错误，该错误非常清楚地表明每个消费者组的每个分区只能有 5 个消费者(或者类似的东西)。

此外，似乎(我们不能 100% 确定)这个问题是在我们重写代码(上面)以便能够为每个分区生成一个线程时开始发生的。 (即使我们在 EventHub 中只有 1 个分区)。 编辑:我们进行了更多的日志挖掘，并且在合并代码以在每个分区生成一个线程之前还发现了一些异常。

Answer 1

该异常表明有另一个使用者配置为使用相同的使用者组并断言对该分区的独占访问。除非您在客户端选项中显式设置 OwnerLevel 属性，否则可能至少有一个 EventProcessorClient 正在运行。

要进行修复，您可以:

• 停止针对同一事件中心和使用者组组合运行的任何事件处理器，并确保没有其他使用者显式设置 OwnerLevel。

• 在专门的消费者组中运行这些消费者；这将使它们能够与独占消费者和/或事件处理器共存。

• 将这些消费者的 OwnerLevel 显式设置为 1 或更高；这将声明所有权并强制同一消费者组中的任何其他消费者断开连接。
  (注意:根据其他使用者的情况，您可能需要在此处测试不同的值。事件处理器类型使用 0，因此高于该值的任何内容都将优先。)