DynamoDB异步GSI查询超时后返回重复数据问题求助

你遇到的重复数据问题，核心原因是**orTimeout仅触发了CompletableFuture的超时回调，但并未终止DynamoDB异步查询的订阅流**。当查询超时后，原查询的SdkPublisher仍然会继续向消费者推送分页数据；后续同一分区重新启动查询时，旧的未终止流和新流同时推送数据，就会导致消费者收到重复的页面（甚至多次重复）。

• 追踪订阅生命周期：给每个查询的订阅添加日志，记录订阅的创建、取消时间，以及超时触发时的状态。确认超时后订阅是否被正确终止，是否还有旧流在推送数据。
• 页面数据标记：在消费者处理页面时，记录每个页面的唯一标识（比如结合分区ID、查询的ExclusiveStartKey、页面序号），同时记录每条数据的主键。通过日志对比重复数据的来源，确认是否来自超时后的旧流。
• 超时场景复现：在测试环境中主动模拟超时（比如缩短PROCESSING_TIMEOUT），观察是否能稳定复现重复数据的情况，同时验证取消订阅后的效果。

1. 超时后强制终止查询订阅 #

DynamoDB Enhanced Async Client的query()返回的SdkPublisher遵循Reactive Streams规范，订阅后会返回Subscription对象，调用其cancel()方法可以终止流的推送。修改代码在超时触发时取消订阅：

修改查询方法，返回订阅对象 #

public CompletableFuture<Subscription> getItemsFromGsiLessThan(Key key, String indexName, Consumer<List<YourItem>> consumer) {
    QueryConditional query = QueryConditional.sortLessThan(key);
    QueryEnhancedRequest request = QueryEnhancedRequest.builder()
            .queryConditional(query)
            .limit(25)
            .build();

    try {
        CompletableFuture<Subscription> future = new CompletableFuture<>();
        SdkPublisher<Page<YourItem>> publisher = table.index(indexName)
                .query(request)
                .limit(5);
        
        // 订阅时获取Subscription对象
        Subscription subscription = publisher.subscribe(
            page -> consumer.accept(page.items()),
            throwable -> future.completeExceptionally(throwable),
            () -> future.complete(null)
        );
        future.complete(subscription);
        return future;
    } catch (Exception e) {
        return CompletableFuture.failedFuture(e);
    }
}

在超时回调中取消订阅 #

Phaser partitionPhaser = new Phaser(1);
partitionPhasers.put(partition, partitionPhaser);
dynamoDbRepository.getItemsFromGsiLessThan(key, gsiName, this)
    .orTimeout(PROCESSING_TIMEOUT, MILLISECONDS)
    .whenComplete((subscription, throwable) -> {
        // 超时触发时取消订阅，终止旧流
        if (throwable instanceof TimeoutException && subscription != null) {
            subscription.cancel();
        }
        
        partitionPhaser.arriveAndAwaitAdvance();
        partitionHandler.notifyStopProcessing(partition);
        partitionPhaser.arriveAndDeregister();
        reportPartitionProcessingTime(partition, startProcessingTime);

        verifyAndHandleException(throwable);
});

2. 消费者侧实现幂等处理 #

即使订阅终止逻辑存在疏漏，也要保证业务逻辑不被重复执行：

• DynamoDB数据标记：处理前先查询数据的状态（比如新增processed字段），通过条件更新标记为已处理，避免重复处理。
• 内存去重：用ConcurrentHashMap记录已处理的主键（注意设置过期时间，避免内存泄漏），处理前先判断主键是否已存在。

3. 优化分区协调逻辑 #

当前Phaser的使用可能存在竞态，比如超时后分区被提前释放，旧消费者仍在运行。调整逻辑确保：

• 只有当所有订阅都被取消、所有消费者处理完成后，才释放分区。
• 在notifyStopProcessing前，先等待所有Phaser的注册者完成（比如调用arriveAndAwaitAdvance()确保所有消费者都已deregister）。

4. 调整查询与超时策略 #

• 适当调大PROCESSING_TIMEOUT，减少不必要的超时触发（比如根据单次查询的平均耗时设置，预留足够的分页处理时间）。
• 优化查询的limit参数，减少分页次数，降低超时概率。如果需要处理大量数据，考虑使用ExclusiveStartKey实现分页的连续性，避免重复查询同一范围。

内容的提问来源于stack exchange，提问作者Cristian