Add CosmosEndToEndOperationLatencyPolicyConfig support to azure-cosmos-benchmark

Add per-request E2E timeout policy and availability strategy configuration
to the benchmark module. The policy is applied on Cosmos(Item|Query)RequestOptions,
not at the client builder level, to avoid impacting metadata calls during startup.

Changes:
- TenantWorkloadConfig: Add endToEndTimeoutMs and availabilityStrategyEnabled
  JSON config fields with getters and applyField support
- AsyncBenchmark: Build CosmosEndToEndOperationLatencyPolicyConfig once in the
  base class constructor, expose as e2ePolicyConfig for subclasses
- AsyncReadBenchmark: Create CosmosItemRequestOptions with E2E policy, pass
  to readItem() (was using the 3-arg overload without options)
- AsyncWriteBenchmark: Create CosmosItemRequestOptions with E2E policy, replace
  null options in createItem()
- AsyncQueryBenchmark: Set E2E policy on CosmosQueryRequestOptions immediately
  after construction (applies to all query operation types)

Usage in workload config JSON:
  "endToEndTimeoutMs": 3000,
  "availabilityStrategyEnabled": true

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jeet1995

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncBenchmark.java

@@ -10,6 +10,8 @@
 import com.azure.cosmos.CosmosAsyncDatabase;
 import com.azure.cosmos.CosmosClientBuilder;
 import com.azure.cosmos.CosmosDiagnosticsThresholds;
+import com.azure.cosmos.CosmosEndToEndOperationLatencyPolicyConfig;
+import com.azure.cosmos.CosmosEndToEndOperationLatencyPolicyConfigBuilder;
 import com.azure.cosmos.CosmosContainerProactiveInitConfigBuilder;
 import com.azure.cosmos.CosmosException;
 import com.azure.cosmos.DirectConnectionConfig;
@@ -60,12 +62,25 @@ abstract class AsyncBenchmark<T> implements Benchmark {
     final String partitionKey;
     final TenantWorkloadConfig workloadConfig;
     final List<PojoizedJson> docsToRead;
+    final CosmosEndToEndOperationLatencyPolicyConfig e2ePolicyConfig;
 
     AsyncBenchmark(TenantWorkloadConfig cfg, Scheduler scheduler) {
 
         workloadConfig = cfg;
         this.benchmarkScheduler = scheduler;
 
+        // Build E2E timeout policy if configured (applied per-request, NOT at client level)
+        if (cfg.getEndToEndTimeoutMs() != null) {
+            logger.info("E2E timeout policy: {}ms (will be set on request options, not client builder)",
+                cfg.getEndToEndTimeoutMs());
+            this.e2ePolicyConfig = new CosmosEndToEndOperationLatencyPolicyConfigBuilder(
+                Duration.ofMillis(cfg.getEndToEndTimeoutMs()))
+                .enable(true)
+                .build();
+        } else {
+            this.e2ePolicyConfig = null;
+        }
+
         final TokenCredential credential = cfg.isManagedIdentityRequired()
             ? cfg.buildTokenCredential()
             : null;

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncQueryBenchmark.java

@@ -41,6 +41,9 @@ protected Mono<FeedResponse<PojoizedJson>> performWorkload(long i) {
         Flux<FeedResponse<PojoizedJson>> obs;
         Random r = new Random();
         CosmosQueryRequestOptions options = new CosmosQueryRequestOptions();
+        if (e2ePolicyConfig != null) {
+            options.setCosmosEndToEndOperationLatencyPolicyConfig(e2ePolicyConfig);
+        }
 
         if (workloadConfig.getOperationType() == Operation.QueryCross) {
 

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncReadBenchmark.java

@@ -3,6 +3,7 @@
 
 package com.azure.cosmos.benchmark;
 
+import com.azure.cosmos.models.CosmosItemRequestOptions;
 import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.PartitionKey;
 
@@ -11,16 +12,22 @@
 
 class AsyncReadBenchmark extends AsyncBenchmark<PojoizedJson> {
 
+    private final CosmosItemRequestOptions readOptions;
+
     AsyncReadBenchmark(TenantWorkloadConfig cfg, Scheduler scheduler) {
         super(cfg, scheduler);
+        this.readOptions = new CosmosItemRequestOptions();
+        if (e2ePolicyConfig != null) {
+            readOptions.setCosmosEndToEndOperationLatencyPolicyConfig(e2ePolicyConfig);
+        }
     }
 
     @Override
     protected Mono<PojoizedJson> performWorkload(long i) {
         int index = (int) (i % docsToRead.size());
         PojoizedJson doc = docsToRead.get(index);
         return cosmosAsyncContainer.readItem(doc.getId(),
-            new PartitionKey(doc.getId()), PojoizedJson.class)
+            new PartitionKey(doc.getId()), readOptions, PojoizedJson.class)
             .map(CosmosItemResponse::getItem);
     }
 }

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/AsyncWriteBenchmark.java

@@ -3,6 +3,7 @@
 
 package com.azure.cosmos.benchmark;
 
+import com.azure.cosmos.models.CosmosItemRequestOptions;
 import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.PartitionKey;
 
@@ -16,12 +17,17 @@ class AsyncWriteBenchmark extends AsyncBenchmark<CosmosItemResponse> {
 
     private final String uuid;
     private final String dataFieldValue;
+    private final CosmosItemRequestOptions writeOptions;
 
     AsyncWriteBenchmark(TenantWorkloadConfig cfg, Scheduler scheduler) {
         super(cfg, scheduler);
 
         uuid = UUID.randomUUID().toString();
         dataFieldValue = RandomStringUtils.randomAlphabetic(workloadConfig.getDocumentDataFieldSize());
+        this.writeOptions = new CosmosItemRequestOptions();
+        if (e2ePolicyConfig != null) {
+            writeOptions.setCosmosEndToEndOperationLatencyPolicyConfig(e2ePolicyConfig);
+        }
     }
 
     @Override
@@ -40,7 +46,7 @@ protected Mono<CosmosItemResponse> performWorkload(long i) {
                 partitionKey,
                 workloadConfig.getDocumentDataFieldCount()),
                 new PartitionKey(id),
-                null);
+                writeOptions);
         }
         // Raw type cast is required because CosmosItemResponse uses wildcard generics
         // that cannot be expressed in the class type parameter without propagating

sdk/cosmos/azure-cosmos-benchmark/src/main/java/com/azure/cosmos/benchmark/TenantWorkloadConfig.java

@@ -188,6 +188,12 @@ public enum Environment {
     @JsonProperty("aggressiveWarmupDuration")
     private String aggressiveWarmupDuration;
 
+    @JsonProperty("endToEndTimeoutMs")
+    private Integer endToEndTimeoutMs;
+
+    @JsonProperty("availabilityStrategyEnabled")
+    private Boolean availabilityStrategyEnabled;
+
     // ======== Connection params ========
 
     @JsonProperty("connectionMode")
@@ -314,6 +320,9 @@ public Duration getAggressiveWarmupDuration() {
         return Duration.parse(aggressiveWarmupDuration);
     }
 
+    public Integer getEndToEndTimeoutMs() { return endToEndTimeoutMs; }
+    public boolean isAvailabilityStrategyEnabled() { return availabilityStrategyEnabled != null && availabilityStrategyEnabled; }
+
     public int getNumberOfCollectionForCtl() { return numberOfCollectionForCtl != null ? numberOfCollectionForCtl : 4; }
     public String getReadWriteQueryReadManyPct() { return readWriteQueryReadManyPct != null ? readWriteQueryReadManyPct : "90,8,1,1"; }
     public int getEncryptedStringFieldCount() { return encryptedStringFieldCount != null ? encryptedStringFieldCount : 1; }
@@ -496,6 +505,10 @@ private void applyField(String key, String value, boolean overwrite) {
                     if (overwrite || proactiveConnectionRegionsCount == null) proactiveConnectionRegionsCount = Integer.parseInt(value); break;
                 case "aggressiveWarmupDuration":
                     if (overwrite || aggressiveWarmupDuration == null) aggressiveWarmupDuration = value; break;
+                case "endToEndTimeoutMs":
+                    if (overwrite || endToEndTimeoutMs == null) endToEndTimeoutMs = Integer.parseInt(value); break;
+                case "availabilityStrategyEnabled":
+                    if (overwrite || availabilityStrategyEnabled == null) availabilityStrategyEnabled = Boolean.parseBoolean(value); break;
                 case "connectionMode":
                     if (overwrite || connectionMode == null) connectionMode = value; break;
                 case "consistencyLevel":