Azure Event Hub Java客户端仅处理新消息问题排查

问题描述 #

我有一个简单的Java Event Hub客户端（仅1个分区）

public static void main(String[] args) throws Exception {
        // Create a blob container client that you use later to build an event processor client to receive and process events
        BlobContainerAsyncClient blobContainerAsyncClient = new BlobContainerClientBuilder()
                .connectionString(storageConnectionString)
                .containerName(storageContainerName)
                .buildAsyncClient();

        // Create a builder object that you will use later to build an event processor client to receive and process events and errors.
        EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
                .connectionString(connectionString, eventHubName)
                .consumerGroup("$default")
                .processEvent(PARTITION_PROCESSOR)
                .processError(ERROR_HANDLER)//.checkpointStore(new SampleCheckpointStore());
                .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient));

        // Use the builder object to create an event processor client
        EventProcessorClient eventProcessorClient = eventProcessorClientBuilder.buildEventProcessorClient();

        System.out.println("Starting event processor");
        eventProcessorClient.start();

        System.out.println("Press enter to stop.");
        System.in.read();

        System.out.println("Stopping event processor");
        eventProcessorClient.stop();
        System.out.println("Event processor stopped.");

        System.out.println("Exiting process");
    }

当我先启动客户端再发送消息时，消息能正常被处理。

但如果先停止客户端，向Event Hub发送消息后再启动客户端，此前发送的消息完全不会被处理，仅处理启动后发送的消息，这是为什么？

此外，若停止客户端后删除Azure Blob Storage中的检查点数据，再启动客户端，Event Hub中已有的消息仍不会被处理，仅处理启动后发送的消息，这又是为什么？

使用的依赖库：

<dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-messaging-eventhubs</artifactId>
            <version>5.12.2</version>
        </dependency>
        <dependency>
            <groupId>com.azure</groupId>
            <artifactId>azure-messaging-eventhubs-checkpointstore-blob</artifactId>
            <version>1.13.0</version>
        </dependency>

已尝试的无效方案：

• 将依赖库版本改为5.10
• 使用内存检查点存储替代Blob存储

问题分析与解决 #

核心原因 #

问题根源在于EventProcessorClient的默认起始位置策略：

1. 当客户端启动时未找到对应消费者组的检查点数据，会默认采用EventPosition.latest()作为消费起始点——也就是只消费客户端启动之后产生的新消息，不会回溯消费已有的历史消息。
2. 你删除Blob中的检查点后，客户端重启时依然因无检查点而触发默认策略，所以还是只会处理启动后的新消息。
3. 关于“停止客户端后发消息再启动不处理”的情况：如果第一次启动时你未手动提交检查点，客户端重启时同样会使用latest起始位置，自然只会处理重启后的新消息；若第一次启动时意外提交了检查点到当时的最新位置，后续启动会从该检查点继续，而你停止后发送的消息若在检查点之后却未被处理，大概率是检查点提交逻辑存在问题（比如未正确调用检查点更新方法）。

解决方案 #

1. 配置初始消费起始位置

在构建EventProcessorClientBuilder时，显式指定无检查点时从事件流的最早位置开始消费：

EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
        .connectionString(connectionString, eventHubName)
        .consumerGroup("$default")
        .processEvent(PARTITION_PROCESSOR)
        .processError(ERROR_HANDLER)
        .checkpointStore(new BlobCheckpointStore(blobContainerAsyncClient))
        // 关键配置：无检查点时从最早位置消费
        .startingPosition(EventPosition.earliest());

2. 强制忽略检查点（可选）

如果希望每次启动都强制消费所有历史消息（忽略已有的检查点），可以在起始位置配置中添加ignoreCheckpoint参数：

.startingPosition(EventPosition.earliest(), true)

第二个参数设为true后，客户端会完全忽略检查点数据，始终从指定的起始位置开始消费。

3. 确保检查点正确提交

检查你的事件处理器PARTITION_PROCESSOR，确保在处理完消息后手动提交检查点，这样客户端重启时才能从上次中断的位置继续消费：

private static final Consumer<ProcessEventContext> PARTITION_PROCESSOR = context -> {
    // 处理业务逻辑
    System.out.printf("处理分区%s的消息：%s%n",
            context.getPartitionContext().getPartitionId(),
            context.getEventData().getBodyAsString());

    // 手动提交检查点，记录当前消费位置
    context.updateCheckpoint();
};

注意：EventProcessorClient不会自动提交检查点，必须手动调用updateCheckpoint()方法。

内容的提问来源于stack exchange，提问作者Dušan Maruščák