Add batch size validation for transactional batches. Refactored calculateTotalSerializedLength to validate 2MB payload limit and return HTTP 413 errors with clear messages when exceeded.

Author: TheovanKraay

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/batch/TransactionalBulkExecutor.java

@@ -286,6 +286,31 @@ private Flux<CosmosBulkOperationResponse<TContext>> executeTransactionalBatch(Co
         String batchTrackingId = UUIDs.nonBlockingRandomUUID().toString();
         PartitionKey partitionKey = cosmosBatch.getPartitionKeyValue();
 
+        // Validate batch payload size before execution
+        int totalSerializedLength = this.calculateTotalSerializedLength(operations);
+        if (totalSerializedLength > this.maxMicroBatchPayloadSizeInBytes) {
+            String errorMessage = String.format(
+                "Transactional batch exceeds maximum payload size: %d bytes (limit: %d bytes), PK: %s",
+                totalSerializedLength,
+                this.maxMicroBatchPayloadSizeInBytes,
+                partitionKey);
+            logger.error("{}, Context: {}", errorMessage, this.operationContextText);
+            
+            // Return error responses for all operations in the batch
+            CosmosException payloadSizeException = BridgeInternal.createCosmosException(
+                HttpConstants.StatusCodes.REQUEST_ENTITY_TOO_LARGE,
+                errorMessage);
+            
+            return Flux.fromIterable(operations)
+                .map(operation -> {
+                    TContext actualContext = this.getActualContext(operation);
+                    return ModelBridgeInternal.createCosmosBulkOperationResponse(
+                        operation,
+                        payloadSizeException,
+                        actualContext);
+                });
+        }
+        
         logDebugOrWarning(
             "Executing transactional batch - {} operations, PK: {}, TrackingId: {}, Context: {}",
             operations.size(),
@@ -402,14 +427,14 @@ private Flux<CosmosBulkOperationResponse<TContext>> executeTransactionalBatch(Co
             });
     }
 
-    private int calculateTotalSerializedLength(AtomicInteger currentTotalSerializedLength, CosmosItemOperation item) {
-        if (item instanceof CosmosItemOperationBase) {
-            return currentTotalSerializedLength.accumulateAndGet(
-                ((CosmosItemOperationBase) item).getSerializedLength(this.effectiveItemSerializer),
-                Integer::sum);
+    private int calculateTotalSerializedLength(List<CosmosItemOperation> operations) {
+        int totalLength = 0;
+        for (CosmosItemOperation operation : operations) {
+            if (operation instanceof CosmosItemOperationBase) {
+                totalLength += ((CosmosItemOperationBase) operation).getSerializedLength(this.effectiveItemSerializer);
+            }
         }
-
-        return currentTotalSerializedLength.get();
+        return totalLength;
     }
 
     private TContext getActualContext(CosmosItemOperation itemOperation) {