Java实现Cosmos Change Feed Listener报错：leaseClient需启用写入内容响应

错误原因 #

该错误是因为Cosmos Change Feed Processor在操作lease容器（创建、更新lease文档）时，要求启用content response on write设置——这个设置控制写入操作后服务器是否返回文档内容，而Change Feed Processor依赖该返回内容完成lease的同步管理，默认情况下未开启该配置。

修复方案 #

你需要在代码中为lease容器的操作显式启用该设置，以下两种方式任选其一：

方式一：为lease容器设置全局默认请求选项 #

修改构造方法中lease容器的初始化逻辑，添加默认请求选项：

@Autowired
public ChangeFeedListener(CosmosAsyncDatabase cosmosAsyncDatabase) {
    this.sourceContainer = cosmosAsyncDatabase.getContainer("sourceContainer");
    this.targetContainer = cosmosAsyncDatabase.getContainer("targetContainer");
    // 为lease容器启用content response on write
    CosmosItemRequestOptions defaultLeaseOptions = new CosmosItemRequestOptions()
            .setContentResponseOnWriteEnabled(true);
    this.leaseContainer = cosmosAsyncDatabase.getContainer("leaseContainer")
            .withDefaultItemRequestOptions(defaultLeaseOptions);
}

方式二：通过ChangeFeedProcessorOptions指定lease操作的请求选项 #

在初始化ChangeFeedProcessorOptions时，专门配置lease相关的请求选项：

@EventListener({ApplicationStartedEvent.class})
public void initialize() {
    // 创建并配置ChangeFeedProcessorOptions
    ChangeFeedProcessorOptions changeFeedProcessorOptions = new ChangeFeedProcessorOptions();
    // 为lease操作启用content response on write
    CosmosItemRequestOptions leaseRequestOptions = new CosmosItemRequestOptions()
            .setContentResponseOnWriteEnabled(true);
    changeFeedProcessorOptions.setLeaseItemRequestOptions(leaseRequestOptions);

    // 构建Change Feed Processor
    ChangeFeedProcessor changeFeedProcessor = new ChangeFeedProcessorBuilder()
            .hostName(UUID.randomUUID().toString())
            .feedContainer(sourceContainer)
            .handleChanges((List<JsonNode> docs) -> {
                for (JsonNode document : docs) {
                    System.out.println("Processing document: " + document);
                    // 建议改用异步处理替代block()，避免线程阻塞
                    targetContainer.createItem(document)
                            .subscribe(
                                    success -> System.out.println("Document written to target container."),
                                    error -> System.err.println("Failed to write document: " + error.getMessage())
                            );
                }
            })
            .leaseContainer(leaseContainer)
            .options(changeFeedProcessorOptions)
            .buildChangeFeedProcessor();

    changeFeedProcessor.start();
}

额外优化提示 #

代码中使用targetContainer.createItem(document).block()会阻塞当前线程，在异步容器场景下可能引发线程池耗尽等问题，建议改用异步订阅的方式处理写入结果，如上述代码中的subscribe示例。

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