[Internal] DTS: Adds required document id to all DTS operations (#5793)

## Description

- The coordinator needs `documentId` to reliably commit or abort stuck transactions—the transaction log doesn’t store document content.
- Always sending it enables deterministic replay without request bodies and avoids special cases (especially for create).
- Making it mandatory keeps the SDK–backend contract future‑proof and correct.

All distributed transaction operations now require an explicit 'id' parameter, consistent with Delete, Replace, and Patch. Previously, Create and Upsert omitted id from the metadata sent to the DTC coordinator (the id was only in the resource body).

- Add string id parameter to CreateItem, CreateItemStream, UpsertItem, UpsertItemStream in both the abstract class and the core implementation
- Validate id at call site via existing ValidateItemId (fails fast with ArgumentNullException before serialization)
- Simplify DistributedTransactionOperation<T>: one constructor (with id), MaterializeResourceAsync just serializes and calls base
- Update all tests

> Note: To ensure consistency with Create/Upsert operations in other SDK flows, we will implement a long-term solution by extracting the id from the resource body instead of having users provide it explicitly.
 
## Type of change

Please delete options that are not relevant.

- [] Bug fix (non-breaking change which fixes an issue)
- [] New feature (non-breaking change which adds functionality)
- [] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [] This change requires a documentation update

## Closing issues

To automatically close an issue: closes #IssueNumber

---------

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

Author: Meghana-Palaparthi

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionOperation.cs

@@ -1,4 +1,4 @@
-// ------------------------------------------------------------
+// ------------------------------------------------------------
 // Copyright (c) Microsoft Corporation.  All rights reserved.
 // ------------------------------------------------------------
 
@@ -78,19 +78,6 @@ internal virtual async Task MaterializeResourceAsync(CosmosSerializerCore serial
 
     internal class DistributedTransactionOperation<T> : DistributedTransactionOperation
     {
-        public DistributedTransactionOperation(
-            Documents.OperationType operationType,
-            int operationIndex,
-            string database,
-            string container,
-            PartitionKey partitionKey,
-            T resource,
-            DistributedTransactionRequestOptions requestOptions = null)
-            : base(operationType, operationIndex, database, container, partitionKey, id: null, requestOptions)
-        {
-            this.Resource = resource;
-        }
-
         public DistributedTransactionOperation(
             Documents.OperationType operationType,
             int operationIndex,

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedTransactionSerializer.cs

@@ -6,6 +6,7 @@ namespace Microsoft.Azure.Cosmos
 {
     using System;
     using System.Collections.Generic;
+    using System.Diagnostics;
     using System.IO;
     using System.Text.Json;
     using Microsoft.Azure.Documents;
@@ -62,6 +63,9 @@ private static void WriteOperation(Utf8JsonWriter jsonWriter, DistributedTransac
             // collectionName
             jsonWriter.WriteString("collectionName", operation.Container);
 
+            // id
+            jsonWriter.WriteString("id", operation.Id);
+
             // collectionResourceId
             if (operation.CollectionResourceId != null)
             {
@@ -74,12 +78,6 @@ private static void WriteOperation(Utf8JsonWriter jsonWriter, DistributedTransac
                 jsonWriter.WriteString("databaseResourceId", operation.DatabaseResourceId);
             }
 
-            // id 
-            if (operation.Id != null)
-            {
-                jsonWriter.WriteString("id", operation.Id);
-            }
-
             // partitionKey
             if (operation.PartitionKeyJson != null)
             {

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedWriteTransaction.cs

@@ -24,13 +24,15 @@ abstract class DistributedWriteTransaction : DistributedTransaction
         /// <param name="database">The name of the database containing the container.</param>
         /// <param name="collection">The name of the container where the item will be created.</param>
         /// <param name="partitionKey">The partition key for the item.</param>
+        /// <param name="id">The unique identifier of the item to create.</param>
         /// <param name="resource">The resource to create.</param>
         /// <param name="requestOptions">Options for the create operation.</param>
         /// <returns>The current <see cref="DistributedWriteTransaction"/> instance for method chaining.</returns>
         public abstract DistributedWriteTransaction CreateItem<T>(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             T resource,
             DistributedTransactionRequestOptions requestOptions = null);
 
@@ -40,13 +42,15 @@ public abstract DistributedWriteTransaction CreateItem<T>(
         /// <param name="database">The name of the database containing the container.</param>
         /// <param name="collection">The name of the container where the item will be created.</param>
         /// <param name="partitionKey">The partition key for the item.</param>
+        /// <param name="id">The unique identifier of the item to create.</param>
         /// <param name="streamPayload">A <see cref="Stream"/> containing the JSON serialization of the item.</param>
         /// <param name="requestOptions">Options for the create operation.</param>
         /// <returns>The current <see cref="DistributedWriteTransaction"/> instance for method chaining.</returns>
         public abstract DistributedWriteTransaction CreateItemStream(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             Stream streamPayload,
             DistributedTransactionRequestOptions requestOptions = null);
 
@@ -147,13 +151,15 @@ public abstract DistributedWriteTransaction PatchItemStream(
         /// <param name="database">The name of the database containing the container.</param>
         /// <param name="collection">The name of the container where the item will be upserted.</param>
         /// <param name="partitionKey">The partition key for the item.</param>
+        /// <param name="id">The unique identifier of the item to upsert.</param>
         /// <param name="resource">The resource to upsert.</param>
         /// <param name="requestOptions">Options for the upsert operation.</param>
         /// <returns>The current <see cref="DistributedWriteTransaction"/> instance for method chaining.</returns>
         public abstract DistributedWriteTransaction UpsertItem<T>(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             T resource,
             DistributedTransactionRequestOptions requestOptions = null);
 
@@ -164,13 +170,15 @@ public abstract DistributedWriteTransaction UpsertItem<T>(
         /// <param name="database">The name of the database containing the container.</param>
         /// <param name="collection">The name of the container where the item will be upserted.</param>
         /// <param name="partitionKey">The partition key for the item.</param>
+        /// <param name="id">The unique identifier of the item to upsert.</param>
         /// <param name="streamPayload">A <see cref="Stream"/> containing the JSON serialization of the item.</param>
         /// <param name="requestOptions">Options for the upsert operation.</param>
         /// <returns>The current <see cref="DistributedWriteTransaction"/> instance for method chaining.</returns>
         public abstract DistributedWriteTransaction UpsertItemStream(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             Stream streamPayload,
             DistributedTransactionRequestOptions requestOptions = null);
     }

Microsoft.Azure.Cosmos/src/DistributedTransaction/DistributedWriteTransactionCore.cs

@@ -27,10 +27,12 @@ public override DistributedWriteTransaction CreateItem<T>(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             T resource,
             DistributedTransactionRequestOptions requestOptions = null)
         {
             DistributedWriteTransactionCore.ValidateContainerReference(database, collection);
+            DistributedWriteTransactionCore.ValidateItemId(id);
             DistributedWriteTransactionCore.ValidateResource(resource);
 
             this.operations.Add(
@@ -40,6 +42,7 @@ public override DistributedWriteTransaction CreateItem<T>(
                     database,
                     collection,
                     partitionKey,
+                    id,
                     resource,
                     requestOptions));
             return this;
@@ -49,10 +52,12 @@ public override DistributedWriteTransaction CreateItemStream(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             Stream streamPayload,
             DistributedTransactionRequestOptions requestOptions = null)
         {
             DistributedWriteTransactionCore.ValidateContainerReference(database, collection);
+            DistributedWriteTransactionCore.ValidateItemId(id);
             if (streamPayload == null)
             {
                 throw new ArgumentNullException(nameof(streamPayload));
@@ -65,6 +70,7 @@ public override DistributedWriteTransaction CreateItemStream(
                     database: database,
                     container: collection,
                     partitionKey: partitionKey,
+                    id: id,
                     requestOptions: requestOptions)
                 {
                     ResourceStream = streamPayload
@@ -214,10 +220,12 @@ public override DistributedWriteTransaction UpsertItem<T>(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             T resource,
             DistributedTransactionRequestOptions requestOptions = null)
         {
             DistributedWriteTransactionCore.ValidateContainerReference(database, collection);
+            DistributedWriteTransactionCore.ValidateItemId(id);
             DistributedWriteTransactionCore.ValidateResource(resource);
 
             this.operations.Add(
@@ -227,6 +235,7 @@ public override DistributedWriteTransaction UpsertItem<T>(
                     database,
                     collection,
                     partitionKey,
+                    id,
                     resource,
                     requestOptions));
             return this;
@@ -236,10 +245,12 @@ public override DistributedWriteTransaction UpsertItemStream(
             string database,
             string collection,
             PartitionKey partitionKey,
+            string id,
             Stream streamPayload,
             DistributedTransactionRequestOptions requestOptions = null)
         {
             DistributedWriteTransactionCore.ValidateContainerReference(database, collection);
+            DistributedWriteTransactionCore.ValidateItemId(id);
             if (streamPayload == null)
             {
                 throw new ArgumentNullException(nameof(streamPayload));
@@ -252,6 +263,7 @@ public override DistributedWriteTransaction UpsertItemStream(
                     database: database,
                     container: collection,
                     partitionKey: partitionKey,
+                    id: id,
                     requestOptions: requestOptions)
                 {
                     ResourceStream = streamPayload