【Event Hub】消费消息时遇到Azure.Messaging.EventHubs.EventHubsException(QuotaExceeded) 错误的解决之法

问题描述

使用EventHubConsumerClient消费Event Hub的消息时，遇见Azure.Messaging.EventHubs.EventHubsException(QuotaExceeded): Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.”异常。

代码片段为：

... 
private static String connectionString = "Event Hub Connection ...";
private static String consumerGroup = "xxxxxx";
private static EventHubConsumerClient consumer =  new EventHubClientBuilder()
           .connectionString(connectionString)
           .consumerGroup(consumerGroup)
           .buildConsumerClient();
...
   while (true) {
       IterableStream<PartitionEvent> events = consumer.receiveFromPartition(pId,100, sPosition, Duration.ofMillis(1000));
       int count = 0;
       for (PartitionEvent partitionEvent : events) {
           EventData event = partitionEvent.getData();
...

错误消息为：

com.azure.core.amqp.exception.AmqpException: Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.

List of connected receivers - ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c, ff7e6a04-43a4-4d50-ae86-e0586297f60c.

问题解答

这是因为Event Hub对于同一个使用者组的限制：建议在使用者组中一个分区上只有一个活动接收器。 但是，在某些情况下，每个分区最多可以使用 5 个使用者或接收器，其中所有接收器都会获取分区的所有事件。

根据文档中提到的非 Epoch Receiver的限制为5，而如果为每一个Receiver设置上Epoch值，当新的Receiver连接到Event Hub服务器后，服务端会保留最大Epoch值的Receiver，其它旧的则会被逐渐删除。

例如下面的代码就可以解决Azure.Messaging.EventHubs.EventHubsException(QuotaExceeded): Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5.”异常。

import com.azure.messaging.eventhubs.models.ReceiveOptions;
... 

private static String connectionString = "Event Hub Connection ...";
private static String consumerGroup = "xxxxxx";

private static EventHubConsumerClient consumer =  new EventHubClientBuilder()
           .connectionString(connectionString)
           .consumerGroup(consumerGroup)
           .buildConsumerClient();

...

   ReceiveOptions receiveOptions ;
   receiveOptions = new ReceiveOptions();
   receiveOptions.setOwnerLevel((new Date()).getTime()); // set epoch value


   while (true) {
       IterableStream<PartitionEvent> events = consumer.receiveFromPartition(pId,100, sPosition, Duration.ofMillis(1000),receiveOptions );
       int count = 0;
       for (PartitionEvent partitionEvent : events) {
           EventData event = partitionEvent.getData();

...

参考资料

1. https://docs.azure.cn/zh-cn/event-hubs/event-hubs-features#consumer-groups
2. https://docs.azure.cn/zh-cn/event-hubs/event-hubs-quotas

当在复杂的环境中面临问题，格物之道需：浊而静之徐清，安以动之徐生。 云中，恰是如此!