调用hasNext()时Cosmos DB读取挂起的原因排查

问题背景 #

代码在Cosmos DB模拟器中运行正常，但生产环境调用hasNext()时出现读取挂起，查询预期仅返回0或1条数据：

CosmosContainer fs = ...;
SqlQuerySpec sql = new SqlQuerySpec( ... );
CosmosPagedIterable<EntryPOJO> items = fs.queryItems( sql, null, EntryPOJO.class );
Iterator<EntryPOJO> it = items.iterator();
if( it.hasNext() ) {
    EntryPOJO entry = it.next();
}

挂起时的堆栈跟踪：

"ForkJoinPool.commonPool-worker-1" #13 daemon prio=5 os_prio=0 cpu=52.86ms elapsed=519.67s tid=0x00007fecfd4e6a40 nid=0x61d waiting on condition  [0x00007fecfccfe000]
   java.lang.Thread.State: WAITING (parking)
        at jdk.internal.misc.Unsafe.park(java.base@17.0.4/Native Method)
        - parking to wait for  <0x0000000080664c98> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(java.base@17.0.4/UnknownSource)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(java.base@17.0.4/Unknown Source)
        at java.util.concurrent.ForkJoinPool.compensatedBlock(java.base@17.0.4/Unknown Source)
        at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.4/Unknown Source)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@17.0.4/Unknown Source)
        at reactor.core.publisher.BlockingIterable$SubscriberIterator.hasNext(BlockingIterable.java:179)

排查原因 #

• 阻塞迭代器的线程依赖问题：CosmosPagedIterable的迭代器基于Reactor的BlockingIterable实现，默认使用ForkJoinPool.commonPool。生产环境中若commonPool线程被其他任务占用、线程数配置不足，会导致await()一直等待信号。
• 未配置查询超时：代码中未设置请求超时时间，生产环境网络波动、Cosmos DB端处理延迟时，线程会无限期等待响应。
• 生产环境连接/重试策略差异：模拟器和生产环境的连接池、重试策略配置不同，生产环境可能因重试策略不合理（如无限制重试）导致请求卡在重试流程。
• 查询性能瓶颈：即便预期返回少量数据，若查询未命中合适索引，生产环境数据量大会导致查询执行时间过长，间接引发挂起。

解决方案 #

• 添加查询超时配置：通过QueryOptions设置明确的超时时间，避免无限等待：

QueryOptions options = new QueryOptions();
options.setRequestTimeout(Duration.ofSeconds(10)); // 根据业务调整超时时间
CosmosPagedIterable<EntryPOJO> items = fs.queryItems(sql, options, EntryPOJO.class);

• 改用非阻塞异步API：避免使用阻塞迭代器，利用Reactor异步模型处理结果，更适合生产环境：

fs.queryItems(sql, null, EntryPOJO.class)
    .singleOrEmpty() // 适配0或1条结果的场景
    .subscribe(
        entry -> { /* 处理查询结果 */ },
        error -> { /* 处理异常情况 */ }
    );

• 指定专用线程池：若必须使用阻塞API，为迭代器指定专用线程池，避免占用commonPool：

ScheduledExecutorService dedicatedExecutor = Executors.newSingleThreadScheduledExecutor();
Iterator<EntryPOJO> it = items.iterator(dedicatedExecutor);
// 使用完成后关闭线程池
dedicatedExecutor.shutdown();

• 优化查询索引：检查查询的WHERE条件字段，确保配置了对应的复合索引或范围索引，提升查询执行效率。
• 调整客户端连接配置：确认生产环境的CosmosClient配置了合理的重试策略和连接池参数：

CosmosClient client = new CosmosClientBuilder()
    .endpoint(yourEndpoint)
    .key(yourKey)
    .connectionPolicy(ConnectionPolicy.getDefaultPolicy())
    .consistencyLevel(ConsistencyLevel.SESSION)
    .retryOptions(new RetryOptions().setMaxRetryAttemptsOnThrottledRequests(5)) // 限制节流重试次数
    .buildClient();

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