Java WebFlux中Azure CosmosAsyncClient密钥轮换重建报错排查

公司政策规定Azure CosmosDB密钥需定期轮换并存储在Azure KeyVault中，需要在Java WebFlux响应式应用中处理密钥变更。由于Java SDK创建CosmosAsyncClient后无法修改密钥，尝试通过以下代码从KeyVault获取新密钥重建客户端：

SecretAsyncClient secretAsyncClient = new SecretClientBuilder().buildAsyncClient();
// ... 省略其他配置
Mono<CosmosAsyncClient> client = secretAsyncClient.getSecret(KEY_NAME)
    .map(Secret::getValue)
    .map(key -> new CosmosClientBuilder()
        .endpoint(HOST)
        .key(key)
        .buildAsyncClient());

return client.flatMap(client -> {
    // ... 业务逻辑
});

日志显示getSecret()执行正常，但出现以下错误：

05:25:00.597 INFO  c.a.c.i.RxDocumentClientImpl - Initializing DocumentClient [4] with serviceEndpoint [https://xxxx-cosmosdb-sql-dev.documents.azure.com:443/], connectionPolicy [ConnectionPolicy{httpNetworkRequestTimeout=PT1M, tcpNetworkRequestTimeout=PT5S, connectionMode=DIRECT, maxConnectionPoolSize=1000, idleHttpConnectionTimeout=PT1M, idleTcpConnectionTimeout=PT0S, userAgentSuffix='', throttlingRetryOptions=RetryOptions{maxRetryAttemptsOnThrottledRequests=9, maxRetryWaitTime=PT30S}, endpointDiscoveryEnabled=true, preferredRegions=[Switzerland North], multipleWriteRegionsEnabled=true, proxyType=null, inetSocketProxyAddress=null, readRequestsFallbackEnabled=true, connectTimeout=PT5S, idleTcpEndpointTimeout=PT1H, maxConnectionsPerEndpoint=130, maxRequestsPerConnection=30, tcpConnectionEndpointRediscoveryEnabled=true}], consistencyLevel [Session], directModeProtocol [Tcp]
05:25:00.598 ERROR c.a.c.i.RxDocumentClientImpl - unexpected failure in initializing client.
java.lang.RuntimeException: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-3
        at com.azure.cosmos.implementation.http.ReactorNettyClient.attemptToWarmupHttpClient(ReactorNettyClient.java:118)
        at com.azure.cosmos.implementation.http.ReactorNettyClient.createWithConnectionProvider(ReactorNettyClient.java:98)
        at com.azure.cosmos.implementation.http.HttpClient.createFixed(HttpClient.java:61)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.httpClient(RxDocumentClientImpl.java:617)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:410)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:262)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:230)
        at com.azure.cosmos.implementation.AsyncDocumentClient$Builder.build(AsyncDocumentClient.java:243)
        at com.azure.cosmos.CosmosAsyncClient.<init>(CosmosAsyncClient.java:129)
        at com.azure.cosmos.CosmosClientBuilder.buildAsyncClient(CosmosClientBuilder.java:779)

想知道除了将CosmosClientBuilder.buildAsyncClient()放入线程池外，有没有更简洁优雅的解决方式？

错误原因分析 #

CosmosClientBuilder.buildAsyncClient()内部初始化环节存在阻塞调用（比如HTTP客户端预热时使用了block()），而WebFlux的默认工作线程（如reactor-http-epoll-*）属于非阻塞线程池，禁止执行阻塞操作，因此触发了上述异常。

优雅的处理方案 #

1. 切换到弹性线程池执行客户端初始化

使用Reactor的subscribeOn(Schedulers.boundedElastic())将客户端创建操作切换到专门处理阻塞任务的弹性线程池，同时添加客户端缓存逻辑避免重复创建（CosmosAsyncClient是重量级对象，频繁重建会消耗资源）：

private static final String HOST = "your-cosmos-endpoint";
private static final String KEY_NAME = "your-keyvault-secret-name";
private final SecretAsyncClient secretAsyncClient;
private volatile Mono<CosmosAsyncClient> cachedClient;

// 构造函数初始化SecretAsyncClient
public CosmosClientManager(SecretAsyncClient secretAsyncClient) {
    this.secretAsyncClient = secretAsyncClient;
    this.cachedClient = refreshClient();
}

private Mono<CosmosAsyncClient> refreshClient() {
    return secretAsyncClient.getSecret(KEY_NAME)
        .map(Secret::getValue)
        // 切换到弹性线程池执行阻塞的客户端初始化
        .subscribeOn(Schedulers.boundedElastic())
        .map(key -> new CosmosClientBuilder()
            .endpoint(HOST)
            .key(key)
            .buildAsyncClient())
        // 缓存客户端实例，按密钥轮换周期设置缓存时长
        .cache(cosmosClient -> Duration.ofHours(24), 
            error -> Duration.ZERO,
            () -> Duration.ZERO);
}

// 对外提供获取客户端的方法，支持主动刷新
public Mono<CosmosAsyncClient> getClient(boolean refresh) {
    if (refresh) {
        cachedClient = refreshClient();
    }
    return cachedClient;
}

2. 利用Azure SDK的密钥自动刷新（推荐）

Azure Cosmos DB Java SDK支持通过TokenCredential结合Azure Identity实现密钥的自动刷新，无需手动从KeyVault获取密钥重建客户端：

// 使用DefaultAzureCredential自动从KeyVault获取并刷新密钥
TokenCredential credential = new DefaultAzureCredentialBuilder()
    .build();

// 创建客户端时使用TokenCredential而非固定密钥
CosmosAsyncClient client = new CosmosClientBuilder()
    .endpoint(HOST)
    .credential(credential)
    .buildAsyncClient();

这种方式下，SDK会自动处理密钥的轮换和刷新，无需手动管理客户端重建，是最简洁的解决方案。

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