[Hotfix] Binary Encoding: Fixes Serialization Gaps on Newtonsoft Reader/Writer for Transactional Batch (#5041)

# Pull Request Template

## Description

This PR cherry-picks the following commits:

- #5030

## Type of change

Please delete options that are not relevant.

- [x] Bug fix (non-breaking change which fixes an issue)

## Closing issues

To automatically close an issue: closes #IssueNumber

Author: kundadebdatta

Directory.Build.props

@@ -1,8 +1,8 @@
 <Project xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
 	<PropertyGroup>
-		<ClientOfficialVersion>3.47.1</ClientOfficialVersion>
+		<ClientOfficialVersion>3.47.2</ClientOfficialVersion>
 		<ClientPreviewVersion>3.48.0</ClientPreviewVersion>
-		<ClientPreviewSuffixVersion>preview.1</ClientPreviewSuffixVersion>
+		<ClientPreviewSuffixVersion>preview.2</ClientPreviewSuffixVersion>
 		<DirectVersion>3.37.10</DirectVersion>
 		<FaultInjectionVersion>1.0.0</FaultInjectionVersion>
 		<FaultInjectionSuffixVersion>beta.0</FaultInjectionSuffixVersion>

Microsoft.Azure.Cosmos/contracts/API_3.47.2.txt

@@ -183,8 +183,14 @@ private static Result ReadOperationResult(ref RowReader reader, out Transactiona
                             return r;
                         }
 
-                        batchOperationResult.ResourceStream = new MemoryStream(
-                            buffer: resourceBody, index: 0, count: resourceBody.Length, writable: false, publiclyVisible: true);
+                        batchOperationResult.ResourceStream = new CloneableStream(
+                            internalStream: new MemoryStream(
+                                buffer: resourceBody,
+                                index: 0,
+                                count: resourceBody.Length,
+                                writable: false,
+                                publiclyVisible: true),
+                            allowUnsafeDataAccess: true);
                         break;
 
                     case "requestCharge":

Microsoft.Azure.Cosmos/contracts/API_3.48.0-preview.2.txt

@@ -182,7 +182,19 @@ public override bool Read()
         /// <returns>A <see cref="byte"/>[] or <c>null</c> if the next JSON token is null. This method will return <c>null</c> at the end of an array.</returns>
         public override byte[] ReadAsBytes()
         {
-            throw new NotImplementedException();
+            this.Read();
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.Null)
+            {
+                return null;
+            }
+
+            if (this.jsonReader.CurrentTokenType == JsonTokenType.EndArray)
+            {
+                return new byte[0];
+            }
+
+            string stringValue = this.jsonReader.GetStringValue();
+            return Convert.FromBase64String(stringValue);
         }
 
         /// <summary>

Microsoft.Azure.Cosmos/src/Batch/TransactionalBatchOperationResult.cs

@@ -331,7 +331,7 @@ public override void WriteValue(DateTime value)
         /// <param name="value">The <see cref="byte"/>[] value to write.</param>
         public override void WriteValue(byte[] value)
         {
-            throw new NotSupportedException("Can not write byte arrays");
+            this.WriteValue(Convert.ToBase64String(value));
         }
 
         /// <summary>

Microsoft.Azure.Cosmos/src/Json/Interop/CosmosDBToNewtonsoftReader.cs

@@ -145,5 +145,93 @@ public async Task ReadMany2UnreachablePartitionsTest()
                 fiClient.Dispose();
             }
         }
+
+        [TestMethod]
+        [Owner("dkunda")]
+        [TestCategory("MultiRegion")]
+        [DataRow(true, DisplayName = "Test scenario when binary encoding is enabled at client level.")]
+        [DataRow(false, DisplayName = "Test scenario when binary encoding is disabled at client level.")]
+        public async Task ExecuteTransactionalBatch_WhenBinaryEncodingEnabled_ShouldCompleteSuccessfully(
+            bool isBinaryEncodingEnabled)
+        {
+            Environment.SetEnvironmentVariable(ConfigurationManager.BinaryEncodingEnabled, isBinaryEncodingEnabled.ToString());
+
+            Random random = new();
+            CosmosIntegrationTestObject testItem = new()
+            {
+                Id = $"smTestId{random.Next()}",
+                Pk = $"smpk{random.Next()}",
+            };
+
+            try
+            {
+                CosmosClientOptions cosmosClientOptions = new()
+                {
+                    ConsistencyLevel = ConsistencyLevel.Session,
+                    RequestTimeout = TimeSpan.FromSeconds(10),
+                    Serializer = new CosmosJsonDotNetSerializer(
+                        cosmosSerializerOptions: new CosmosSerializationOptions()
+                        {
+                            PropertyNamingPolicy = CosmosPropertyNamingPolicy.CamelCase
+                        },
+                        binaryEncodingEnabled: isBinaryEncodingEnabled)
+                };
+
+                using CosmosClient cosmosClient = new(
+                    connectionString: this.connectionString,
+                    clientOptions: cosmosClientOptions);
+
+                Database database = cosmosClient.GetDatabase(MultiRegionSetupHelpers.dbName);
+                Container container = database.GetContainer(MultiRegionSetupHelpers.containerName);
+
+                // Create a transactional batch
+                TransactionalBatch transactionalBatch = container.CreateTransactionalBatch(new PartitionKey(testItem.Pk));
+
+                transactionalBatch.CreateItem(
+                    testItem,
+                    new TransactionalBatchItemRequestOptions
+                    {
+                        EnableContentResponseOnWrite = true,
+                    });
+
+                transactionalBatch.ReadItem(
+                    testItem.Id,
+                    new TransactionalBatchItemRequestOptions
+                    {
+                        EnableContentResponseOnWrite = true,
+                    });
+
+                // Execute the transactional batch
+                TransactionalBatchResponse transactionResponse = await transactionalBatch.ExecuteAsync(
+                    new TransactionalBatchRequestOptions
+                    {
+                    });
+
+                Assert.AreEqual(HttpStatusCode.OK, transactionResponse.StatusCode);
+                Assert.AreEqual(2, transactionResponse.Count);
+
+                TransactionalBatchOperationResult<CosmosIntegrationTestObject> createOperationResult = transactionResponse.GetOperationResultAtIndex<CosmosIntegrationTestObject>(0);
+                
+                Assert.IsNotNull(createOperationResult);
+                Assert.IsNotNull(createOperationResult.Resource);
+                Assert.AreEqual(testItem.Id, createOperationResult.Resource.Id);
+                Assert.AreEqual(testItem.Pk, createOperationResult.Resource.Pk);
+
+                TransactionalBatchOperationResult<CosmosIntegrationTestObject> readOperationResult = transactionResponse.GetOperationResultAtIndex<CosmosIntegrationTestObject>(1);
+                
+                Assert.IsNotNull(readOperationResult);
+                Assert.IsNotNull(readOperationResult.Resource);
+                Assert.AreEqual(testItem.Id, readOperationResult.Resource.Id);
+                Assert.AreEqual(testItem.Pk, readOperationResult.Resource.Pk);
+            }
+            finally
+            {
+                Environment.SetEnvironmentVariable(ConfigurationManager.BinaryEncodingEnabled, null);
+
+                await this.container.DeleteItemAsync<CosmosIntegrationTestObject>(
+                    testItem.Id,
+                    new PartitionKey(testItem.Pk));
+            }
+        }
     }
 }

Microsoft.Azure.Cosmos/src/Json/Interop/CosmosDBToNewtonsoftWriter.cs

@@ -129,7 +129,9 @@ public async Task BatchResponseDeserializationAsync()
                 new TransactionalBatchOperationResult(HttpStatusCode.Conflict),
                 new TransactionalBatchOperationResult(HttpStatusCode.OK)
                 {
-                    ResourceStream = new MemoryStream(new byte[] { 0x41, 0x42 }, index: 0, count: 2, writable: false, publiclyVisible: true),
+                    ResourceStream = new CloneableStream(
+                        internalStream: new MemoryStream(new byte[] { 0x41, 0x42 }, index: 0, count: 2, writable: false, publiclyVisible: true),
+                        allowUnsafeDataAccess: true),
                     RequestCharge = 2.5,
                     ETag = "1234",
                     RetryAfter = TimeSpan.FromMilliseconds(360)
@@ -250,7 +252,7 @@ private bool Equals(Stream x, Stream y)
                         return false;
                     }
 
-                    return ((MemoryStream)x).GetBuffer().SequenceEqual(((MemoryStream)y).GetBuffer());
+                    return ((CloneableStream)x).GetBuffer().SequenceEqual(((CloneableStream)y).GetBuffer());
                 }
 
                 return false;