Azure 事件中心 Java 客户端未处理数据答案

【问题标题】：Data not being processed by Azure Event Hub Java clientAzure 事件中心 Java 客户端未处理数据
【发布时间】：2018-06-20 17:04:25
【问题描述】：

在EventProcessorHost example 之后，我们在 onEvents() 中实现了我们的自定义逻辑。有些数据没有被处理，我怀疑这是因为 Java 客户端抛出的警告。

在日志中，我们看到 StorageException（用于更新租约或检查点的 Blob 存储超时）、LeaseLostException（可能是由于之前的异常）和 EventHubException（当事件中心移动或短时间脱机时）。

基本上我的问题是：这些异常如何影响事件的处理以及我们如何确保没有事件被跳过（例如，通过重试的异常处理和作为最后手段完全关闭）？

我通读了docs 并搜索了其他无法找到满意答案的问题（this 和this 提供了一些见解）。

我们的代码：

public class EventProcessor implements IEventProcessor {
    ...
    @Override
    public void onEvents(PartitionContext context, Iterable<EventData> events) throws Exception {
        for (EventData event : events) {
            try {
                String message = new String(event.getBytes(), StandardCharsets.UTF_8);

                mystuff.process(message);

                this.checkpointBatchingCount++;
                if ((checkpointBatchingCount % 50) == 0) {
                    context.checkpoint(data).get();
                }
            } catch (Exception e) {
                LOG.warn("Processing event failed: {}", e.getMessage())
            }
        }
    }
    ...
}

【问题讨论】：

我一直在寻找您的确切问题，想知道您是否最终找到了解决方案。我查看了作为 eventthubs 包 (com.microsoft.azure.eventhubs.*) 一部分的 PartitionReceiveHandler，它包含一个 onReceive，我不确定它是否能解决手头的问题。似乎 IEventProcessor 不能用作侦听器；在调用 registerEventProcessor 时，仅在所需的 onEvents() 方法中获取到该时间点的事件；然后它挂起，没有错误或退出，并且要从所需的 EventHub 获取更多消息，我们必须重新
老实说，我对 Azure Event Hub Java 客户端非常失望，我们遇到了一个又一个问题（在他们的 Github 存储库中查看我的问题）。我很高兴我当时工作的公司完全脱离了 Azure Event Hub，并在稍后阶段迁移到另一个流媒体平台。

标签： java azure azure-eventhub

【解决方案1】：

根据我对 EventProcessor 的理解，您将重新处理事件而不是丢失事件。可能还有另一个潜在问题。

当您调用 checkpoint 时会发生什么，它会保留该 EventData 的序列号（偏移量等）流，并表示“我已经处理了这个”。

当您收到StorageException 时，这意味着序列号未成功持久化，因此旧事件的序列号仍存在于您的 Blob 存储中。如果遇到EventHubException 处理器在重新启动时断开连接的情况，它将尝试声明任何已过期的租约并从最后一个成功的检查点开始处理。

如果另一个事件处理器“窃取”了您当前正在处理的分区，您将获得 LeaseLostException。当有多个 EventProcessor 实例正在运行并且客户端尝试平衡正在运行的实例之间的分区数量时，就会发生这种情况。

【讨论】：