[Cosmos] Reduce Change Feed Processor lease write response overhead

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -15,6 +15,7 @@
 
 * Updated Netty to `4.2.19.Final`, Reactor Netty to `1.3.7`, and Reactor Core to `3.8.7`. Migrated direct TCP event loops to the Netty 4.2 I/O handler API while preserving pooled buffers and explicit TLS validation settings.
 * Added a compact `ppaf` bookmark to each data-plane attempt in `CosmosDiagnostics`, containing the current per-partition write region, failed regions, and the time it was designated, or an empty object when no override is active.
+* Change Feed Processor lease writes no longer request the document in the response, reducing response bandwidth and client response-processing overhead without changing RU charges. The new lease ETag is taken from the response headers, and the lease container's client no longer needs `contentResponseOnWriteEnabled(true)`. - See [PR 50697](https://github.com/Azure/azure-sdk-for-java/pull/50697)
 
 ### 4.83.0 (2026-09-22)
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/common/ChangeFeedHelper.java

@@ -2,6 +2,8 @@
 // Licensed under the MIT License.
 package com.azure.cosmos.implementation.changefeed.common;
 
+import com.azure.cosmos.models.CosmosItemRequestOptions;
+
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -24,6 +26,20 @@ public class ChangeFeedHelper {
     public static final int HTTP_STATUS_CODE_TOO_MANY_REQUESTS = 429;
     public static final int HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR = 500;
 
+    /**
+     * Disables content response on write for a lease container write request. Lease state is preserved locally
+     * and only the new ETag (concurrency token) is taken from the response headers, so the response body is
+     * never needed. This overrides the content-response-on-write setting of the lease client.
+     *
+     * @param options the request options to update; a new instance is created when null.
+     * @return the request options with content response on write disabled.
+     */
+    public static CosmosItemRequestOptions withContentResponseOnWriteDisabled(CosmosItemRequestOptions options) {
+        CosmosItemRequestOptions effectiveOptions = options != null ? options : new CosmosItemRequestOptions();
+        effectiveOptions.setContentResponseOnWriteEnabled(false);
+        return effectiveOptions;
+    }
+
     public static String getDatabaseLink(String databaseName) {
         return String.format("/dbs/%s", databaseName);
     }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/ChangeFeedProcessorImplBase.java

@@ -145,10 +145,6 @@ private void validateChangeFeedProcessorOptions(ChangeFeedProcessorOptions chang
     private void validateLeaseContainer(CosmosAsyncContainer leaseContainer) {
         checkNotNull(leaseContainer, "Argument 'leaseContainer' can not be null");
 
-        if (!getContextClient(leaseContainer).isContentResponseOnWriteEnabled()) {
-            throw new IllegalArgumentException("leaseClient: content response on write setting must be enabled");
-        }
-
         ConsistencyLevel consistencyLevel = getContextClient(leaseContainer).getConsistencyLevel();
         if (consistencyLevel == ConsistencyLevel.CONSISTENT_PREFIX || consistencyLevel == ConsistencyLevel.EVENTUAL) {
             logger.warn("leaseClient consistency level setting are less then expected which is SESSION");

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/DocumentServiceLeaseUpdaterImpl.java

@@ -11,7 +11,10 @@
 import com.azure.cosmos.implementation.changefeed.ServiceItemLeaseUpdater;
 import com.azure.cosmos.implementation.changefeed.exceptions.LeaseConflictException;
 import com.azure.cosmos.implementation.changefeed.exceptions.LeaseLostException;
+import com.azure.cosmos.implementation.apachecommons.lang.StringUtils;
+import com.azure.cosmos.implementation.changefeed.common.ChangeFeedHelper;
 import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.PartitionKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,10 +64,7 @@ public Mono<Lease> updateLease(
         return
             Mono.just(this)
             .flatMap( value -> this.tryReplaceLease(cachedLease, itemId, partitionKey, requestOptions))
-            .map(leaseDocument -> {
-                cachedLease.setServiceItemLease(ServiceItemLeaseV1.fromDocument(leaseDocument));
-                return cachedLease;
-            })
+            .map(replaceResponse -> this.applyReplaceResponse(cachedLease, replaceResponse))
             .hasElement()
             .flatMap(hasItems -> {
                 if (hasItems) {
@@ -134,7 +134,29 @@ public Mono<Lease> updateLease(
             });
     }
 
-    private Mono<InternalObjectNode> tryReplaceLease(
+    private Lease applyReplaceResponse(Lease cachedLease, CosmosItemResponse<Lease> replaceResponse) {
+        InternalObjectNode leaseDocument = BridgeInternal.getProperties(replaceResponse);
+        if (leaseDocument != null) {
+            cachedLease.setServiceItemLease(ServiceItemLeaseV1.fromDocument(leaseDocument));
+            return cachedLease;
+        }
+
+        // Lease writes are issued with content response on write disabled - the submitted lease state is what got
+        // persisted, so only the new concurrency token needs to be taken from the response headers. Not doing so
+        // would leave a stale If-Match ETag on the cached lease and cause the next update to fail with 412.
+        String newETag = replaceResponse.getETag();
+        if (StringUtils.isEmpty(newETag)) {
+            logger.warn(
+                "Lease with token {}: Replace succeeded but the response has no ETag; the next update will refresh the lease.",
+                cachedLease.getLeaseToken());
+        } else {
+            cachedLease.setConcurrencyToken(newETag);
+        }
+
+        return cachedLease;
+    }
+
+    private Mono<CosmosItemResponse<Lease>> tryReplaceLease(
             Lease lease,
             String itemId,
             PartitionKey partitionKey,
@@ -144,7 +166,6 @@ private Mono<InternalObjectNode> tryReplaceLease(
                 partitionKey,
                 lease,
                 this.getCreateIfMatchOptions(cosmosItemRequestOptions, lease))
-            .map(cosmosItemResponse -> BridgeInternal.getProperties(cosmosItemResponse))
             .onErrorResume(re -> {
                 if (re instanceof CosmosException) {
                     CosmosException ex = (CosmosException) re;
@@ -168,8 +189,10 @@ private Mono<InternalObjectNode> tryReplaceLease(
     }
 
     private CosmosItemRequestOptions getCreateIfMatchOptions(CosmosItemRequestOptions createIfMatchOptions, Lease lease) {
-        createIfMatchOptions.setIfMatchETag(lease.getConcurrencyToken());
+        CosmosItemRequestOptions effectiveOptions =
+            ChangeFeedHelper.withContentResponseOnWriteDisabled(createIfMatchOptions);
+        effectiveOptions.setIfMatchETag(lease.getConcurrencyToken());
 
-        return createIfMatchOptions;
+        return effectiveOptions;
     }
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/LeaseStoreImpl.java

@@ -77,7 +77,8 @@ public Mono<Boolean> markInitialized() {
         return this.client.createItem(
             this.leaseCollectionLink,
                 containerDocument,
-                this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument)),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument))),
                 false)
             .map( item -> {
                 return true;
@@ -108,15 +109,14 @@ public Mono<Boolean> acquireInitializationLock(Duration lockExpirationTime) {
         return this.client.createItem(
             this.leaseCollectionLink,
                 containerDocument,
-                this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument)),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument))),
                 false)
             .map(documentResourceResponse -> {
-                if (BridgeInternal.getProperties(documentResourceResponse) != null) {
-                    this.lockETag = BridgeInternal.getProperties(documentResourceResponse).getETag();
-                    return true;
-                } else {
-                    return false;
-                }
+                // The lock is created with content response on write disabled, so the ETag comes from the
+                // response headers - a successful create always means the lock was acquired.
+                this.lockETag = documentResourceResponse.getETag();
+                return true;
             })
             .onErrorResume(throwable -> {
                 if (throwable instanceof CosmosException) {

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/LeaseStoreManagerImpl.java

@@ -180,7 +180,8 @@ public Mono<Lease> createLeaseIfNotExist(FeedRangeEpkImpl feedRange, String cont
         return this.leaseDocumentClient.createItem(
             this.settings.getLeaseCollectionLink(),
                 documentServiceLease,
-                this.requestOptionsFactory.createItemRequestOptions(documentServiceLease),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(documentServiceLease)),
                 false)
             .onErrorResume( ex -> {
                 if (ex instanceof CosmosException) {
@@ -204,11 +205,16 @@ public Mono<Lease> createLeaseIfNotExist(FeedRangeEpkImpl feedRange, String cont
                 }
 
                 InternalObjectNode document = BridgeInternal.getProperties(documentResourceResponse);
+                if (document != null) {
+                    return documentServiceLease
+                        .withId(document.getId())
+                        .withETag(document.getETag())
+                        .withTs(document.getString(Constants.Properties.LAST_MODIFIED));
+                }
 
-                return documentServiceLease
-                    .withId(document.getId())
-                    .withETag(document.getETag())
-                    .withTs(document.getString(Constants.Properties.LAST_MODIFIED));
+                // Lease writes are issued with content response on write disabled - the submitted lease is what got
+                // persisted, so only the concurrency token needs to be taken from the response headers.
+                return documentServiceLease.withETag(documentResourceResponse.getETag());
             });
     }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/DocumentServiceLeaseUpdaterImpl.java

@@ -5,7 +5,10 @@
 import com.azure.cosmos.BridgeInternal;
 import com.azure.cosmos.CosmosException;
 import com.azure.cosmos.implementation.InternalObjectNode;
+import com.azure.cosmos.implementation.apachecommons.lang.StringUtils;
+import com.azure.cosmos.implementation.changefeed.common.ChangeFeedHelper;
 import com.azure.cosmos.models.CosmosItemRequestOptions;
+import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.PartitionKey;
 import com.azure.cosmos.implementation.changefeed.ChangeFeedContextClient;
 import com.azure.cosmos.implementation.changefeed.Lease;
@@ -56,10 +59,7 @@ public Mono<Lease> updateLease(final Lease cachedLease, String itemId, Partition
         return
             Mono.just(this)
             .flatMap( value -> this.tryReplaceLease(cachedLease, itemId, partitionKey, requestOptions))
-            .map(leaseDocument -> {
-                cachedLease.setServiceItemLease(ServiceItemLease.fromDocument(leaseDocument));
-                return cachedLease;
-            })
+            .map(replaceResponse -> this.applyReplaceResponse(cachedLease, replaceResponse))
             .hasElement()
             .flatMap(hasItems -> {
                 if (hasItems) {
@@ -127,13 +127,34 @@ public Mono<Lease> updateLease(final Lease cachedLease, String itemId, Partition
             });
     }
 
-    private Mono<InternalObjectNode> tryReplaceLease(
+    private Lease applyReplaceResponse(Lease cachedLease, CosmosItemResponse<Lease> replaceResponse) {
+        InternalObjectNode leaseDocument = BridgeInternal.getProperties(replaceResponse);
+        if (leaseDocument != null) {
+            cachedLease.setServiceItemLease(ServiceItemLease.fromDocument(leaseDocument));
+            return cachedLease;
+        }
+
+        // Lease writes are issued with content response on write disabled - the submitted lease state is what got
+        // persisted, so only the new concurrency token needs to be taken from the response headers. Not doing so
+        // would leave a stale If-Match ETag on the cached lease and cause the next update to fail with 412.
+        String newETag = replaceResponse.getETag();
+        if (StringUtils.isEmpty(newETag)) {
+            logger.warn(
+                "Lease with token {}: Replace succeeded but the response has no ETag; the next update will refresh the lease.",
+                cachedLease.getLeaseToken());
+        } else {
+            cachedLease.setConcurrencyToken(newETag);
+        }
+
+        return cachedLease;
+    }
+
+    private Mono<CosmosItemResponse<Lease>> tryReplaceLease(
         Lease lease,
         String itemId,
         PartitionKey partitionKey,
         CosmosItemRequestOptions cosmosItemRequestOptions) throws LeaseLostException {
         return this.client.replaceItem(itemId, partitionKey, lease, this.getCreateIfMatchOptions(cosmosItemRequestOptions, lease))
-            .map(cosmosItemResponse -> BridgeInternal.getProperties(cosmosItemResponse))
             .onErrorResume(re -> {
                 if (re instanceof CosmosException) {
                     CosmosException ex = (CosmosException) re;
@@ -157,7 +178,10 @@ private Mono<InternalObjectNode> tryReplaceLease(
     }
 
     private CosmosItemRequestOptions getCreateIfMatchOptions(CosmosItemRequestOptions createIfMatchOptions, Lease lease) {
-        createIfMatchOptions.setIfMatchETag(lease.getConcurrencyToken());
-        return createIfMatchOptions;
+        CosmosItemRequestOptions effectiveOptions =
+            ChangeFeedHelper.withContentResponseOnWriteDisabled(createIfMatchOptions);
+        effectiveOptions.setIfMatchETag(lease.getConcurrencyToken());
+
+        return effectiveOptions;
     }
 }

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/IncrementalChangeFeedProcessorImpl.java

@@ -135,10 +135,6 @@ private void validateChangeFeedProcessorOptions(ChangeFeedProcessorOptions chang
     private void validateLeaseContainer(CosmosAsyncContainer leaseContainer) {
         checkNotNull(leaseContainer, "Argument 'leaseContainer' can not be null");
 
-        if (!getContextClient(leaseContainer).isContentResponseOnWriteEnabled()) {
-            throw new IllegalArgumentException("leaseClient: content response on write setting must be enabled");
-        }
-
         ConsistencyLevel consistencyLevel = getContextClient(leaseContainer).getConsistencyLevel();
         if (consistencyLevel == ConsistencyLevel.CONSISTENT_PREFIX || consistencyLevel == ConsistencyLevel.EVENTUAL) {
             logger.warn("leaseClient consistency level setting are less then expected which is SESSION");

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/LeaseStoreImpl.java

@@ -77,7 +77,8 @@ public Mono<Boolean> markInitialized() {
         return this.client.createItem(
             this.leaseCollectionLink,
                 containerDocument,
-                this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument)),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument))),
                 false)
             .map( item -> true)
             .onErrorResume(throwable -> {
@@ -106,15 +107,14 @@ public Mono<Boolean> acquireInitializationLock(Duration lockExpirationTime) {
         return this.client.createItem(
             this.leaseCollectionLink,
                 containerDocument,
-                this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument)),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(ServiceItemLeaseV1.fromDocument(containerDocument))),
                 false)
             .map(documentResourceResponse -> {
-                if (BridgeInternal.getProperties(documentResourceResponse) != null) {
-                    this.lockETag = BridgeInternal.getProperties(documentResourceResponse).getETag();
-                    return true;
-                } else {
-                    return false;
-                }
+                // The lock is created with content response on write disabled, so the ETag comes from the
+                // response headers - a successful create always means the lock was acquired.
+                this.lockETag = documentResourceResponse.getETag();
+                return true;
             })
             .onErrorResume(throwable -> {
                 if (throwable instanceof CosmosException) {

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/LeaseStoreManagerImpl.java

@@ -176,7 +176,8 @@ public Mono<Lease> createLeaseIfNotExist(String leaseToken, String continuationT
         return this.leaseDocumentClient.createItem(
             this.settings.getLeaseCollectionLink(),
                 documentServiceLease,
-                this.requestOptionsFactory.createItemRequestOptions(documentServiceLease),
+                ChangeFeedHelper.withContentResponseOnWriteDisabled(
+                    this.requestOptionsFactory.createItemRequestOptions(documentServiceLease)),
                 false)
             .onErrorResume( ex -> {
                 if (ex instanceof CosmosException) {
@@ -200,11 +201,16 @@ public Mono<Lease> createLeaseIfNotExist(String leaseToken, String continuationT
                 }
 
                 InternalObjectNode document = BridgeInternal.getProperties(documentResourceResponse);
+                if (document != null) {
+                    return documentServiceLease
+                        .withId(document.getId())
+                        .withETag(document.getETag())
+                        .withTs(document.getString(Constants.Properties.LAST_MODIFIED));
+                }
 
-                return documentServiceLease
-                    .withId(document.getId())
-                    .withETag(document.getETag())
-                    .withTs(document.getString(Constants.Properties.LAST_MODIFIED));
+                // Lease writes are issued with content response on write disabled - the submitted lease is what got
+                // persisted, so only the concurrency token needs to be taken from the response headers.
+                return documentServiceLease.withETag(documentResourceResponse.getETag());
             });
     }